#!/usr/bin/env python3 """common.py -- the strata bench harness's shared pieces: the arm table, the memory plan, records, launch-once, the lock. Internal (names the laptop and local paths). Read HARNESS-NOTES.md first; PREREG.md carries the registered gates. WHAT LIVES HERE * ARMS: the five registered arms (RECON-v1.md lines 444-450), L-IQ3-t8 (PREREG A3), the quality-only fa-off variants L-IQ3-fa0 and L-Q2-fa0 (PREREG A4) and the optional S-HEAD slot, each with its engine, file, contexts, registered band (PREREG section 2) and memory plan (MemoryMax and the MemAvailable floor, with the arithmetic written out in `memory_plan`). `phase_refusal`, `arm_label` and `g1_llama_side` read the A4 keys; PREREG A5-2 moved the fa-off rows to an 8K context (`"ctx": 8192` in their `llama_flags`) and `fa_off_condition` prints their condition from each attempt's own argv. PREREG A7 adds the llama.cpp controls L-IQ3-ub (`-b 4096 -ub 4096`, its own fit, speed only) and L-IQ3-t16 (16 threads, 4K, optional), no band and no gate, each with "control_of" naming L-IQ3-t8, beside which every table prints it (`control_arms`). PREREG A8 (drafted) adds the exploratory S-NEW slot (Strata v0.1.39): unregistered, no band, no gate ("gates" False), captioned `exploratory (v0.1.39)` (`arm_label`), its engine's environment named in its row ("engine_env", engines.engine_env_problem), and it launches only once PREREG carries A8 (`amendment_refusal`) and only on its pinned commit and binary (`strata_pin_problem`). * PREREG A5-3 / A5-5, the clean speed re-reads: `amendment_stamp` reads A5's registered stamp from PREREG.md, `reread_state` sorts an A5 arm's speed attempts into overlapped (RESULT, started before the stamp) and clean (the first RESULT started at or after it), `Records.launch_once(prereg=...)` opens exactly one more speed attempt for each of the four arms, and `quiet_pass` reads the quiet check's record (quiet.py) the driver requires first. * Records: one directory per (arm, phase, attempt); every JSONL row carries a `row_id` the summary cites. Markers: RESULT.json (the phase ran to its end), REFUSED.json (the engine was launched and G0 refused it: the registered "recorded as refused"), STOPPED.json (a stop condition fired). A refusal BEFORE any engine launch (lock busy, memory below the floor, PREREG drift, the egress probe) is PREFLIGHT-REFUSED.json: the arm was never attempted. * Launch-once: a RESULT or REFUSED marker for (arm, phase) in ANY attempt refuses every later launch; a STOPPED or PREFLIGHT-REFUSED attempt refuses its own attempt number, and only an explicit `--attempt N+1` opens a new one. In the fit and load phases each attempt stands alone (one candidate N, one boot): only its own markers count. * The card lock (~/CARD-LOCK, protocol " "): taken with the word `strata-bench` before any card step, refused if another word holds it, released at the end only if it still names us. """ import datetime as dt import hashlib import json import os import re import subprocess import time HERE = os.path.dirname(os.path.abspath(__file__)) BENCH = os.path.dirname(HERE) # .../plans-2026-10-02/strata-bench VENDOR = os.path.join(HERE, "vendor") RECON = "/workshop/bench-archive/plans-2026-09-29/strata-recon/RECON-v1.md" RECON_GATE_LINES = (461, 492) # the block PREREG.md section 2 copies byte for byte PREREG = os.path.join(BENCH, "PREREG.md") LOCK_WORD = "strata-bench" DEFAULT_LOCK = os.path.expanduser("~/CARD-LOCK") ESTATE = "/workshop/estate" #: The estate commit every house file the harness reads from git is pinned to (origin/master when this was written). ESTATE_COMMIT = "7d2a37bca77eb9669864fc3b71f7f4836c222a26" # ------------------------------------------------------------------------------------------------ the files MODEL_ROOT = "/data/strata-bench/models/ISTA-ed59f920" ISTA_REV = "ed59f92082b1e93c0e96d60a8b11aab089b52f09" #: The pinned sha256 of each file (RECON-v1.md lines 325-329; the lead's fetch unit checks the same values). SHA = {"IQ3_S": {1: "4c1eb2ceb4915e1192f4f386021897bde56a97f40a0bb78bb86465e0f7d2aca3", 2: "316b46f3a2dbd68c900f43136ab9449f9dcc3725dfd8c794847c204bc161e113"}, "Q2_0": {1: "69820c02ec7d0b45ef2ebb19d6620299db749fe2aded7f39f93c6b88b199b720", 2: "316b46f3a2dbd68c900f43136ab9449f9dcc3725dfd8c794847c204bc161e113"}, # AUD-C (PREREG section 6, AUD-1): ISTA's Flash-Next Coder at 5348543e0147355ac9cbcb031184a3546350988e. Shard 1 # from the Hugging Face tree read 2026-10-02T20:09:36Z (aud-arms-receipts/hf-coder-tree.json); shard 2 is the # same LFS object as the ISTA file's shard 2, hard-linked (fetch_receipt reads "LINKED IQ1_M shard 2"). "IQ1_M": {1: "e11083ba855e7666b48ea3f2db6a9c3a20c18751a012cc24f948de91b7087fad", 2: "316b46f3a2dbd68c900f43136ab9449f9dcc3725dfd8c794847c204bc161e113"}} FETCH_LOG = os.path.join(BENCH, "receipts", "fetch-ista.log") STRATA_COMMIT = "8fc40dde5e8897a097ec0ea6ea27f9313d171268" LLAMA_COMMIT = "3cf03257f219afbe7334045ff7c6a06ac68c627d" def shard_path(quant, i, root=MODEL_ROOT): return os.path.join(root, quant, f"Qwen3.8-Flash-Next-GSQ-RCO-{quant}-0000{i}-of-00002.gguf") # ------------------------------------------------------------------------------------------------ the arms CONTEXTS = {"4K": 4000, "32K": 32000, "128K": 128000} # slice-text tokens; see prompts.py #: Strata's own expert-arena sizes (setup.py:64-77 "arena_gb", GB = 1e9 bytes) and setup's RAM rule (ram_gb). #: IQ1_M is ISTA's Coder (AUD-C, PREREG section 6 AUD-1 / AUD-ARMS Part D): setup's coder family, arena 23.4 GB, #: RAM rule 32 GB (strata-delta-receipts/head-files/setup.py:113-115, 158-165). ARENA_GB = {"IQ3_S": 50.29, "Q2_0": 33.97, "IQ1_M": 23.4} SETUP_RAM_GB = {"IQ3_S": 62, "Q2_0": 48, "IQ1_M": 32} GIB = 2 ** 30 ARMS = { "L-IQ3": {"order": 1, "engine": "llama.cpp", "quant": "IQ3_S", "mtp": None, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "S-IQ3", "band": (20, 55), "registered": True}, "S-IQ3": {"order": 2, "engine": "strata", "quant": "IQ3_S", "mtp": True, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "L-IQ3", "band": (40, 90), "registered": True}, "S-IQ3-noMTP": {"order": 3, "engine": "strata", "quant": "IQ3_S", "mtp": False, "contexts": ["4K", "32K"], "quality": False, "pair": None, "band": (25, 55), "registered": True}, "S-Q2": {"order": 4, "engine": "strata", "quant": "Q2_0", "mtp": True, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "L-Q2", "band": (60, 130), "registered": True}, "L-Q2": {"order": 5, "engine": "llama.cpp", "quant": "Q2_0", "mtp": None, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "S-Q2", "band": (10, 45), "registered": True}, # NOT registered: no band, no gate of its own; runs only if the lead's delta read of 8fc40dde..HEAD holds. "S-HEAD": {"order": 6, "engine": "strata", "quant": "IQ3_S", "mtp": True, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "L-IQ3", "band": None, "registered": False}, # PREREG A3 (AUDIT-HARNESS I-1, ruled (b), ledger D-20261002-048): L-IQ3's line at -t 8 -tb 8. Every llama.cpp # launch now carries those flags (engines.llama_argv), so this arm differs from L-IQ3 only in that L-IQ3's speed # attempt a1 ran at llama.cpp's default 4 threads before A3. It reuses L-IQ3's frozen --n-cpu-moe ("ncmoe_from": # the fit is about VRAM, not threads) and L-IQ3's band (registered before any row). Speed only: its quality phase # would repeat L-IQ3's own quality phase, which runs at -t 8 -tb 8 under A3; G1's llama side stays L-IQ3 (until # PREREG A4 below moved it to L-IQ3-fa0). "L-IQ3-t8": {"order": 1.5, "engine": "llama.cpp", "quant": "IQ3_S", "mtp": None, "contexts": ["4K", "32K", "128K"], "quality": False, "pair": "S-IQ3", "band": (20, 55), "registered": True, "ncmoe_from": "L-IQ3"}, # PREREG A7 (ledger D-20261004-095; page A's Q-5, D-20261003-064): the llama.cpp controls, each printed straight # after L-IQ3-t8 ("order", "control_of") with its condition read from each attempt's own argv # (engines.control_condition). L-IQ3-ub is L-IQ3-t8's line plus -b 4096 -ub 4096 ("llama_flags"; -b moves with -ub # because 3cf03257 clamps the micro-batch to the logical batch, default 2048). Its compute buffer grows with the # micro-batch, so it fits its own --n-cpu-moe (no "ncmoe_from": FROZEN-NCMOE-L-IQ3-ub.json) under L-IQ3's fit rule, # then runs one speed phase at 4K/32K/128K. A control: no band, no gate, no quality and no load phase ("phases"). # L-IQ3-t16 (A7-6, Q-5b, optional) is L-IQ3-t8's line at -t 16 -tb 16, 4K only, on L-IQ3's N (threads do not move # VRAM), likewise no band and no gate. "L-IQ3-ub": {"order": 1.55, "engine": "llama.cpp", "quant": "IQ3_S", "mtp": None, "contexts": ["4K", "32K", "128K"], "quality": False, "pair": None, "band": None, "registered": True, "phases": ["fit", "speed"], "llama_flags": {"batch": 4096, "ubatch": 4096}, "control_of": "L-IQ3-t8", "registered_by": "PREREG A7"}, "L-IQ3-t16": {"order": 1.56, "engine": "llama.cpp", "quant": "IQ3_S", "mtp": None, "contexts": ["4K"], "quality": False, "pair": None, "band": None, "registered": True, "ncmoe_from": "L-IQ3", "phases": ["speed"], "llama_flags": {"threads": 16}, "control_of": "L-IQ3-t8", "optional": True, "registered_by": "PREREG A7-6, Q-5b, optional"}, # PREREG A4 (the lead's ruling, ledger D-20261002-056): the fa-off variants. L-IQ3's quality attempts a1 and a2 # both STOPPED on the doorman task's second item with "CUDA error: an illegal memory access" (ggml-cuda.cu:108) and # a kernel Xid 31 MMU fault (XID31-FINDING.md); llama.cpp 3cf03257 is "CUDA: enable sparse fa for qwen4". Each # variant is its parent's line with flash attention off ("llama_flags"), on the parent's frozen --n-cpu-moe # ("ncmoe_from"). `-fa off` forces `-ctv f16`: 3cf03257 refuses a quantized V cache without flash attention # (src/llama-context.cpp:3732-3740; engines.llama_line_problem); the K cache stays q8_0. Quality, plus a load phase # (a boot and a hold, no request: whether the fa-off line fits the card costs one load attempt, never a quality # attempt); never a speed phase ("phases"), so no band and no speed contexts. G1's llama.cpp side for the # parent's pair is read from these rows ("g1_side_for"), and every table prints them with their "label". # PREREG A5-2 (the lead's ruling, ledger D-20261003-063): both fa-off loads at A4's line (-c 131072) STOPPED before # /health on "cudaMalloc failed: out of memory" for the compute buffer (8304.57 MiB at N=34, 8847.57 MiB at N=24; # records/L-IQ3-fa0/load/a1, records/L-Q2-fa0/load/a1), so the fa-off variants run at a context of 8192 # ("ctx" in "llama_flags"): the quality prompts are at most about 512 + 8 tokens, so nothing is cut. "L-IQ3-fa0": {"order": 1.7, "engine": "llama.cpp", "quant": "IQ3_S", "mtp": None, "contexts": [], "quality": True, "pair": "S-IQ3", "band": None, "registered": True, "ncmoe_from": "L-IQ3", "phases": ["quality", "load"], "llama_flags": {"flash_attn": "off", "cache_type_v": "f16", "ctx": 8192}, "label": "fa-off", "g1_side_for": "L-IQ3", "registered_by": "PREREG A4"}, "L-Q2-fa0": {"order": 5.5, "engine": "llama.cpp", "quant": "Q2_0", "mtp": None, "contexts": [], "quality": True, "pair": "S-Q2", "band": None, "registered": True, "ncmoe_from": "L-Q2", "phases": ["quality", "load"], "llama_flags": {"flash_attn": "off", "cache_type_v": "f16", "ctx": 8192}, "label": "fa-off", "g1_side_for": "L-Q2", "registered_by": "PREREG A4"}, # NOT registered (PREREG Amendment A8, drafted in A8-DRAFT.md; STRATA-DELTA-2 C-15): Strata v0.1.39 at 6f32ec07, # exploratory, straight after S-HEAD (6.5). S-HEAD's line: the engine arguments setup writes for this box, with no # --pool-workers (C-11), on the staged build (STAGE-NEW.md: binary 2f315b46..., pack and MTP runtime byte-identical # to S-HEAD's), and STRATA_NO_LARGEPAGES=1 handed to the engine through the config's "env" ("engine_env", H-5), so # the expert arena keeps the 4 KB pages S-HEAD ran on (C-12). No band and no gate: "gates" False makes its quality # row print its counts and its agreement with its pair's G1 side, never a G1, G2 or "Better" verdict; "caption" # puts "exploratory (v0.1.39)" beside its name everywhere the summary prints it (arm_label). It launches only once # PREREG.md carries Amendment A8 ("requires_amendment", amendment_refusal), only on its pinned commit and binary # ("strata_commit", "strata_exe_sha256", strata_pin_problem) and only with --allow-unregistered. Its 32K and 128K # rows print C-13's note ("ctx_notes"): v0.1.39 reads long prompts through #583's byte-sized ring, so a comparison # with S-HEAD there compares two prompt paths. "S-NEW": {"order": 6.5, "engine": "strata", "quant": "IQ3_S", "mtp": True, "contexts": ["4K", "32K", "128K"], "quality": True, "pair": "L-IQ3", "band": None, "registered": False, "gates": False, "caption": "exploratory (v0.1.39)", "requires_amendment": "A8", "strata_commit": "6f32ec070f23ced9f50e704d854d775da52591ab", "strata_exe_sha256": "2f315b46fc3b46d0532abc258b5fac66e1fc9efa1a980cf9a83afb00db9ec6c8", "engine_env": {"STRATA_NO_LARGEPAGES": "1"}, "registered_by": "PREREG A8", "ctx_notes": {"32K": "prompt path differs from S-HEAD (#583 ring, v0.1.39)", "128K": "prompt path differs from S-HEAD (#583 ring, v0.1.39)"}}, } #: `load` (AUDIT-HARNESS B-2): boot, probe, G0 and a short hold with no requests, so a launch slip costs one load #: attempt, never a registered speed or quality arm. A Strata arm's speed and quality runs refuse without a RESULT load. PHASES = ("speed", "quality", "load") #: Phases where each attempt stands alone (one candidate N, one boot): only the attempt's own markers count. #: `aud-smoke` (PREREG section 6, AUD-2 C3): AUD-C's smoke boot is its own phase, so its RESULT never trips the measured #: `aud` phase's launch-once marker; like fit and load, only the smoke attempt's own markers count. PER_ATTEMPT_PHASES = ("fit", "load", "aud-smoke") def ncmoe_arm(arm): """The arm whose FROZEN-NCMOE-.json this arm reads (its own, unless it reuses another arm's fit).""" return ARMS[arm].get("ncmoe_from") or arm def phase_refusal(arm, phase, cfg=None): """None when the arm may run `phase`, else the refusal sentence. An arm row that names "phases" runs only those (PREREG A4: the fa-off variants never run a speed phase). `cfg` is the arm's row (default ARMS[arm]; a derived row, such as an AUD arm's, passes its own).""" cfg = ARMS[arm] if cfg is None else cfg only = cfg.get("phases") if only is None or phase in only: return None return (f"{arm} runs only its {' and '.join(only)} phases ({cfg.get('registered_by') or 'the arm table'}): a " f"{phase} phase is refused") def arm_label(arm): """The arm's name as every table prints it: `L-IQ3-fa0 (fa-off)` for a labelled variant (PREREG A4), followed by ` · ` for an arm whose every figure carries one (PREREG A8: `S-NEW · exploratory (v0.1.39)`), else its name.""" row = ARMS.get(arm) or {} lab, cap = row.get("label"), row.get("caption") name = f"{arm} ({lab})" if lab else arm return f"{name} · {cap}" if cap else name def amendment_refusal(arm, prereg=None, cfg=None): """None when the arm may launch as far as its registration goes, else the refusal sentence: an arm row that names "requires_amendment" (PREREG A8 for S-NEW) launches only once `prereg` carries that amendment's registered heading (`amendment_stamp`), so the harness can hold a drafted arm without it ever running before its line is registered. An arm that names none is unaffected.""" cfg = ARMS[arm] if cfg is None else cfg tag = cfg.get("requires_amendment") if not tag: return None prereg = prereg or PREREG if amendment_stamp(prereg, tag): return None return (f"{arm} launches only once PREREG.md carries Amendment {tag}, which registers its line " f"({cfg.get('registered_by') or 'the arm table'}); {prereg} carries no `## N · Amendment {tag} (registered " f"` heading") def strata_pin_problem(cfg, commit, exe): """PREREG A8: the Strata line an arm row pins, checked before any launch. None when the row pins nothing (every arm but S-NEW); else {ok, why, commit_pinned, commit_given, exe, exe_sha256, exe_sha256_pinned}: the --strata-commit given must be the row's "strata_commit" (prefix-wise, at least 7 hex digits) and the engine binary's sha256 the row's "strata_exe_sha256". build_identity then reads the source tree's HEAD against that same commit.""" pin_c, pin_x = cfg.get("strata_commit"), cfg.get("strata_exe_sha256") if not pin_c and not pin_x: return None rec = {"commit_pinned": pin_c, "commit_given": commit, "exe": exe, "exe_sha256_pinned": pin_x, "exe_sha256": None} why = [] if pin_c: c = str(commit or "") if len(c) < 7 or not (c.startswith(pin_c) or pin_c.startswith(c)): why.append(f"--strata-commit {commit!r} is not the commit the arm row pins, {pin_c}") if pin_x: try: rec["exe_sha256"] = sha256_file(exe) except OSError as e: why.append(f"the engine binary {exe} cannot be hashed ({type(e).__name__})") else: if rec["exe_sha256"] != pin_x: why.append(f"the engine binary {exe} reads sha256 {rec['exe_sha256'][:16]}…, not the pinned " f"{pin_x[:16]}…") rec["ok"] = not why rec["why"] = "; ".join(why) or "the commit and the engine binary are the ones the arm row pins" return rec def g1_llama_side(arm): """The arm whose quality rows stand as `arm`'s side of G1 (PREREG A4: the variant whose "g1_side_for" names it), else `arm` itself.""" return next((k for k, v in ARMS.items() if v.get("g1_side_for") == arm), arm) def control_arms(base=None): """PREREG A7: the control arms (rows that name "control_of"), in table order; only `base`'s when given.""" return [k for k, v in sorted(ARMS.items(), key=lambda kv: kv[1]["order"]) if v.get("control_of") and (base is None or v["control_of"] == base)] def _argv_flag(argv, name): argv = list(argv or []) return argv[argv.index(name) + 1] if name in argv[:-1] else None def fa_off_condition(argv): """PREREG A5-1 / A5-2: an fa-off attempt's condition as every fa-off table prints it, read from that attempt's own argv (never from the arm row, so A4's two 128K loads print their own context): `fa-off (V cache f16, forced) at an 8K context` for `-fa off -ctv f16 -c 8192`. None when the argv does not name `-fa off`.""" if _argv_flag(argv, "-fa") != "off": return None v, c = _argv_flag(argv, "-ctv") or "f16", _argv_flag(argv, "-c") if c is not None and c.isdigit() and int(c) > 0 and int(c) % 1024 == 0: k = f"{int(c) // 1024}K" where = f"at {'an' if k[:1] == '8' or k[:2] in ('11', '18') else 'a'} {k} context" else: where = f"at a context of {c} tokens" if c is not None else "at an unread context (no -c in the argv)" return f"fa-off (V cache {v}, forced) {where}" def memory_plan(arm, n_cpu_moe=None, mem_total_gib=None): """{memory_max_gib, need_gib, swap_max, arithmetic}: the unit's MemoryMax and the MemAvailable floor below which the arm is refused. Every number is derived here and the sentence that derives it rides every record.""" return memory_plan_for(ARMS[arm]["engine"], ARMS[arm]["quant"], n_cpu_moe, mem_total_gib) def memory_plan_for(engine, q, n_cpu_moe=None, mem_total_gib=None): """memory_plan's arithmetic for an (engine, quant) pair, so an arm that takes another arm's line on a file chosen at run time (the AUD arms, PREREG section 6) is planned by the same sentence.""" arena_gib = ARENA_GB[q] * 1e9 / GIB total = mem_total_gib if mem_total_gib is not None else read_meminfo().get("MemTotal", 0) / GIB if engine == "strata": # Strata pins the whole expert arena in host RAM (34-50 GB, RECON "How it works"); beside it: the CUDA host # context (~1 GiB), 6 conversation checkpoints x ~118 MB (generate.cpp:280-283), the PLE row cache # (1,048,576 x 90 B), the server and its tokenizer (~0.5 GiB) and the MTP layer's host share (~0.7 GiB): # ~3 GiB, rounded up. With --kv-resident (Q2_0 at 128K by setup's rule) the KV cache joins it in pinned RAM: # 131072 x 13 x 1056 B (setup.py:1814) = 1.80 GB. kv_res = kv_resident_on(q, total) kv_gib = (131072 * 13 * 1056) / GIB if kv_res else 0.0 resident = arena_gib + 3.0 + kv_gib mmax = round(resident + 4.0) need = round(resident + 1.5, 1) text = (f"Strata {q}: arena {ARENA_GB[q]} GB = {arena_gib:.2f} GiB pinned + 3.0 GiB host overhead" + (f" + {kv_gib:.2f} GiB KV in pinned RAM (--kv-resident)" if kv_res else "") + f" = {resident:.2f} GiB resident; MemoryMax = resident + 4 GiB (page cache of the shard read at start " f"is reclaimable inside the cap) = {mmax} GiB; refuse below MemAvailable {need} GiB (resident + 1.5 " f"GiB for the box). Box MemTotal {total:.2f} GiB leaves {total - mmax:.2f} GiB outside the cap.") else: # llama.cpp mmaps shard 1 (-lm mmap) and keeps the first N layers' experts on the CPU (--n-cpu-moe N): those # pages must stay in the page cache charged to the unit or every token re-reads them from the SSD. The # n-gram table (shard 2, 28.8 GB) is read on demand (--lazy-mode on): its touched pages, ~64 KiB per token # worst case (RECON physics), stay small. Host compute buffers and the server: ~3 GiB. n = 48 if n_cpu_moe is None else int(n_cpu_moe) cpu_exp = arena_gib * n / 48.0 resident = cpu_exp + 3.0 mmax = round(min(resident + 14.0, total - 6.0)) need = round(resident + 4.0, 1) text = (f"llama.cpp {q}: CPU experts {n}/48 x {arena_gib:.2f} GiB = {cpu_exp:.2f} GiB of mmapped pages " f"(--n-cpu-moe {n_cpu_moe if n_cpu_moe is not None else 'UNSET: worst case 48'}) + 3.0 GiB host " f"buffers = {resident:.2f} GiB working set; MemoryMax = min(working set + 14 GiB of page-cache room " f"for the GPU layers' pages and the n-gram rows, MemTotal - 6) = {mmax} GiB (a page-cache " f"reclaim at the cap shows as SSD reads/s, recorded per run, never as a kill); refuse below " f"MemAvailable {need} GiB (working set + 4 GiB).") return {"memory_max_gib": mmax, "need_gib": need, "swap_max": "0", "arithmetic": text} def kv_resident_on(quant, mem_total_gib): """setup.py:1820's rule at a 131072 context, int8 KV: ram >= ram_gb + kv_ram_gb + 1 (ram = MemTotal in GiB, setup.py:199-205; kv_ram_gb = ctx x 13 x 1056 / 1e9).""" kv_ram_gb = 131072 * 13 * 1056 / 1e9 return mem_total_gib >= SETUP_RAM_GB[quant] + kv_ram_gb + 1 #: AUDIT-HARNESS B-1: two token counts of one Strata reply disagree when their gap is above 3 % of the reported count #: (8 of 256 tokens). One rule for the driver's run-time comparison and the summary's recount of the kept text. COUNT_TOLERANCE_FRAC = 0.03 def count_gap_too_big(reported, other): """True when |reported - other| is above COUNT_TOLERANCE_FRAC of the reported count; None when either is unread.""" if reported is None or other is None: return None return abs(int(reported) - int(other)) > COUNT_TOLERANCE_FRAC * int(reported) # ------------------------------------------------------------------------------------------------ small tools def utc(t=None): return dt.datetime.fromtimestamp(time.time() if t is None else t, dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def sha256_file(path, chunk=1 << 22): h = hashlib.sha256() with open(path, "rb") as fh: while True: b = fh.read(chunk) if not b: break h.update(b) return h.hexdigest() def sha256_bytes(b): return hashlib.sha256(b).hexdigest() def read_meminfo(path="/proc/meminfo"): out = {} try: for line in open(path, encoding="utf-8"): k, v = line.split(":", 1) f = v.split() out[k] = int(f[0]) * (1024 if len(f) > 1 and f[1] == "kB" else 1) except OSError: pass return out def write_json(path, obj): os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as fh: json.dump(obj, fh, indent=1, sort_keys=False) fh.write("\n") os.replace(tmp, path) def read_json(path, default=None): try: with open(path, encoding="utf-8") as fh: return json.load(fh) except (OSError, ValueError): return default def append_jsonl(path, obj): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "a", encoding="utf-8") as fh: fh.write(json.dumps(obj, sort_keys=False) + "\n") def load_jsonl(path): out = [] try: for line in open(path, encoding="utf-8"): line = line.strip() if line: out.append(json.loads(line)) except OSError: pass return out def git_show(path, commit=ESTATE_COMMIT, repo=ESTATE): """A house file's bytes at the pinned commit, read from git (never a fetch, never a working-tree read).""" return subprocess.run(["git", "-C", repo, "show", f"{commit}:{path}"], capture_output=True, check=True).stdout # ------------------------------------------------------------------------------------------------ PREREG def prereg_check(prereg=PREREG, recon=RECON): """(ok, detail): PREREG.md exists, carries a registered stamp, and its section-2 block equals RECON-v1.md lines 461-492 byte for byte. The harness refuses to start a run otherwise.""" try: p = open(prereg, encoding="utf-8").read() r = open(recon, encoding="utf-8").read().split("\n") except OSError as e: return False, {"why": f"unreadable: {e}"} block = "\n".join(r[RECON_GATE_LINES[0] - 1:RECON_GATE_LINES[1]]) stamp = re.search(r"\*\*Registered (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z)\*\*", p) ok = block in p and stamp is not None return ok, {"prereg_sha256": sha256_bytes(p.encode()), "recon_sha256": sha256_bytes("\n".join(r).encode()), "registered": stamp.group(1) if stamp else None, "block_verbatim": block in p, "why": None if ok else ("the gate block is not verbatim" if stamp else "no registered stamp")} # ------------------------------------------------------------------------------------------------ records MARKERS = ("RESULT.json", "REFUSED.json", "STOPPED.json", "PREFLIGHT-REFUSED.json") class Records: """records///a/: markers, rows.jsonl, telemetry, logs.""" def __init__(self, root, arm, phase, attempt=1): self.root, self.arm, self.phase, self.attempt = root, arm, phase, int(attempt) self.phase_dir = os.path.join(root, arm, phase) self.dir = os.path.join(self.phase_dir, f"a{self.attempt}") self._seq = {} def attempts(self): try: return sorted(int(d[1:]) for d in os.listdir(self.phase_dir) if re.fullmatch(r"a\d+", d)) except OSError: return [] def markers(self, attempt=None): d = os.path.join(self.phase_dir, f"a{attempt}") if attempt is not None else self.dir return [m for m in MARKERS if os.path.exists(os.path.join(d, m))] def launch_once(self, prereg=None): """None when this attempt may launch, else the refusal sentence (section 2's launch-once). A fit attempt is one candidate --n-cpu-moe and a load attempt is one boot, so in those phases only the attempt's own markers count. With `prereg` (the driver passes its PREREG.md), an A5 arm's speed phase whose RESULT attempt a lane's dry run overlapped follows PREREG A5-5 instead (`reread_launch`); without it, or for any other (arm, phase), nothing changes.""" if prereg is not None: st = reread_state(self.root, self.arm, self.phase, amendment_stamp(prereg)) if st is not None: return self.reread_launch(st) for a in ([] if self.phase in PER_ATTEMPT_PHASES else self.attempts()): ms = self.markers(a) for m in ("RESULT.json", "REFUSED.json"): if m in ms: return (f"LAUNCH-ONCE: {self.arm} {self.phase} already has {m} in attempt a{a}; an arm whose " f"RESULT or REFUSED record exists is never re-run") ms = self.markers() if ms: return (f"LAUNCH-ONCE: {self.arm} {self.phase} attempt a{self.attempt} already has {', '.join(ms)}; " f"a new attempt needs an explicit --attempt {max(self.attempts()) + 1}") if os.path.isdir(self.dir) and os.listdir(self.dir): return (f"LAUNCH-ONCE: {self.arm} {self.phase} attempt a{self.attempt} has files but no marker (a run " f"died mid-way?); read {self.dir}, then open --attempt {max(self.attempts()) + 1}") return None def reread_launch(self, st): """PREREG A5-5's launch-once for an A5 arm's speed phase (`st` from reread_state): exactly one more attempt, numbered one past the last overlapped one, may launch once; a REFUSED in any attempt still refuses every launch.""" head = f"LAUNCH-ONCE: {self.arm} {self.phase}" if st["refused"]: return (f"{head} already has REFUSED.json in attempt a{st['refused'][0]}; an arm whose RESULT or REFUSED " f"record exists is never re-run") if st["unplaced"]: return (f"{head}: RESULT attempt a{st['unplaced'][0]} carries no driver_started, so PREREG A5-3 cannot " f"place it before or after A5's stamp ({st['stamp']}); the lead reads it first") if st["after"]: return (f"{head} already has its clean re-read (PREREG A5-3): RESULT.json in attempt a{st['after'][0]}, " f"started at or after A5's stamp {st['stamp']}; it is the figure of record and is never re-run") if self.attempt in st["overlapped"]: return (f"{head} attempt a{self.attempt} already has RESULT.json: it overlapped a lane's dry run and stands " f"as recorded (PREREG A5-3); the one clean re-read is attempt a{st['reread']}") if self.attempt != st["reread"]: return (f"{head}: PREREG A5-3 opens exactly one clean re-read, attempt a{st['reread']} (one past the " f"overlapped a{max(st['overlapped'])}); attempt a{self.attempt} is refused") ms = self.markers() if ms: return (f"{head} attempt a{self.attempt} (the clean re-read PREREG A5-3 registers) already has " f"{', '.join(ms)}; A5-3 registers one re-read, so another attempt needs a new amendment") if os.path.isdir(self.dir) and os.listdir(self.dir): return (f"{head} attempt a{self.attempt} has files but no marker (a run died mid-way?); read {self.dir}: " f"A5-3 registers one re-read, so another attempt needs a new amendment") return None def refuse_launch_once(self, sentence): """A launch-once refusal is logged beside the records, never written as a marker (it ran nothing).""" os.makedirs(self.phase_dir, exist_ok=True) with open(os.path.join(self.phase_dir, "LAUNCH-ONCE-REFUSALS.log"), "a", encoding="utf-8") as fh: fh.write(f"{utc()} {sentence}\n") def path(self, name): return os.path.join(self.dir, name) def row_id(self, kind): n = self._seq[kind] = self._seq.get(kind, 0) + 1 return f"{self.arm}/{self.phase}/a{self.attempt}/{kind}-{n:03d}" def mark(self, marker, obj): assert marker in MARKERS, marker obj = dict(obj, marker=marker.replace(".json", ""), arm=self.arm, phase=self.phase, attempt=self.attempt, written=utc()) write_json(self.path(marker), obj) return obj # ------------------------------------------------------------------------------------------------ PREREG A5 #: PREREG A5-3 (the lead's ruling, ledger D-20261003-063): a lane's dry run overlapped these arms' speed attempts. Each #: overlapped attempt stands as recorded; each arm gets ONE more speed attempt in a quiet window (quiet.py), the clean #: re-read, which is its figure of record whatever its direction (A5-5: the first RESULT speed attempt whose driver #: started at or after A5's stamp). Until it exists the summary reads the overlapped attempt, labelled pending. A5_REREAD_ARMS = ("S-Q2", "S-IQ3", "S-HEAD", "L-Q2") A5_REREAD_PHASE = "speed" OVERLAPPED_LABEL = "overlapped a lane's dry run" CLEAN_PENDING_LABEL = "clean re-read pending" #: quiet.py writes one line per check here, in the arm's phase folder (beside the attempts, never inside one). QUIET_FILE = "QUIET-CHECK.jsonl" #: A5-5: the driver starts a clean re-read only on a quiet-check PASS for that attempt that ended this recently. QUIET_MAX_AGE_S = 900 RE_STAMP = r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z" def amendment_stamp(prereg=PREREG, tag="A5"): """The registered stamp in PREREG's `## N · Amendment (registered ` heading, or None (no such section, or the file unreadable).""" try: txt = open(prereg, encoding="utf-8").read() except OSError: return None m = re.search(r"^## \d+ · Amendment " + re.escape(tag) + r" \(registered (" + RE_STAMP + r")", txt, re.M) return m.group(1) if m else None def reread_state(records, arm, phase, stamp): """PREREG A5-3 / A5-5 for one (arm, phase), or None when the re-read does not apply: not one of the four arms' speed phase, A5 not registered (`stamp` None), or no RESULT attempt that started before A5's stamp. Otherwise {stamp, attempts: {n: {marker, started, ended}}, overlapped, after, unplaced, refused, reread, clean}: overlapped = RESULT attempts whose driver started before the stamp; after = RESULT attempts started at or after it (the first is the clean re-read, `clean`); unplaced = RESULT attempts with no driver_started; refused = REFUSED attempts; reread = the one attempt number A5-5 opens (one past the last overlapped attempt).""" if arm not in A5_REREAD_ARMS or phase != A5_REREAD_PHASE or not stamp: return None r = Records(records, arm, phase) att = {} for n in r.attempts(): ms = r.markers(n) body = read_json(os.path.join(r.phase_dir, f"a{n}", ms[0])) if ms else None att[n] = {"marker": ms[0][:-5] if ms else None, "started": ((body or {}).get("facts") or {}).get("driver_started"), "ended": (body or {}).get("driver_ended")} res = sorted(n for n, x in att.items() if x["marker"] == "RESULT") overlapped = [n for n in res if att[n]["started"] and att[n]["started"] < stamp] if not overlapped: return None after = [n for n in res if att[n]["started"] and att[n]["started"] >= stamp] return {"stamp": stamp, "attempts": att, "overlapped": overlapped, "after": after, "unplaced": [n for n in res if not att[n]["started"]], "refused": sorted(n for n, x in att.items() if x["marker"] == "REFUSED"), "reread": max(overlapped) + 1, "clean": after[0] if after else None} def quiet_lines(records, arm, phase, attempt=None): """The quiet check's record lines (quiet.py) for (arm, phase), oldest first; only `attempt`'s when given.""" rows = load_jsonl(os.path.join(records, arm, phase, QUIET_FILE)) return [x for x in rows if attempt is None or x.get("attempt") == attempt] def quiet_pass(records, arm, phase, attempt, max_age_s=QUIET_MAX_AGE_S, now=None): """(ok, detail): PREREG A5-5, the driver's half of the quiet check. ok when the NEWEST quiet-check line for this attempt reads PASS and ended at most `max_age_s` ago; detail is that line, else the sentence saying why not.""" rows = quiet_lines(records, arm, phase, attempt) if not rows: return False, (f"no quiet check for {arm} {phase} a{attempt} in {arm}/{phase}/{QUIET_FILE}: run quiet.py " f"first (PREREG A5-3)") last = rows[-1] if last.get("verdict") != "PASS": return False, (f"the newest quiet check for {arm} {phase} a{attempt} reads {last.get('verdict')} (ended " f"{last.get('utc_end')}): {str(last.get('sentence'))[:160]}") t_end = last.get("t_end") age = (time.time() if now is None else now) - (t_end if isinstance(t_end, (int, float)) else 0) if age > max_age_s or age < -60: return False, (f"the quiet check's PASS for {arm} {phase} a{attempt} ended {last.get('utc_end')}, {age:.0f} s " f"ago, outside {max_age_s} s: run quiet.py again") return True, last # ------------------------------------------------------------------------------------------------ the lock class CardLock: """~/CARD-LOCK = " \\n" (the house protocol's " " plus a per-run token, so two runs of this bench that both carry the word strata-bench never release each other's lock). take() refuses if another word holds it; release() removes it only if it still carries this run's token.""" def __init__(self, path=DEFAULT_LOCK, word=LOCK_WORD): import secrets self.path, self.word = path, word self.token = secrets.token_hex(6) def holder(self): try: txt = open(self.path, encoding="utf-8").read().strip() except FileNotFoundError: return None return txt.split()[0] if txt else "" def take(self): """(ok, sentence).""" try: fd = os.open(self.path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644) with os.fdopen(fd, "w") as fh: fh.write(f"{self.word} {utc()} {self.token}\n") return True, f"LOCK TAKEN: {self.path} now names {self.word} (token {self.token})" except FileExistsError: h = self.holder() if h == self.word: return False, (f"LOCK REFUSED: {self.path} already names {self.word} from another run " f"({open(self.path).read().strip()!r}); if no strata-bench run is alive, the lead " f"removes it by hand") return False, f"LOCK REFUSED: {self.path} is held by {h!r}, not {self.word!r}" def release(self): try: f = open(self.path, encoding="utf-8").read().split() except FileNotFoundError: return f"LOCK GONE: {self.path} was already absent" if len(f) >= 3 and f[0] == self.word and f[2] == self.token: os.unlink(self.path) return f"LOCK RELEASED: {self.path} (token {self.token})" return f"LOCK LEFT: {self.path} reads {' '.join(f)!r}, not this run's token; not touched" # ------------------------------------------------------------------------------------------------ the fetch receipt def fetch_receipt(quant, log=FETCH_LOG): """{shard: {sha256, bytes, line}} from the lead's fetch log's OK lines (or {} when absent).""" out = {} try: lines = open(log, encoding="utf-8").read().splitlines() except OSError: return out for ln in lines: m = re.match(r"^(\S+) OK (\S+) bytes=(\d+) secs=\d+ sha256=([0-9a-f]{64})$", ln) if m and m.group(2).startswith(f"{quant}/"): i = 1 if "-00001-of-" in m.group(2) else 2 out[i] = {"sha256": m.group(4), "bytes": int(m.group(3)), "at": m.group(1), "line": ln} if quant != "IQ3_S" and 2 not in out: # shard 2 is hard-linked from the IQ3_S copy (Q2_0, and iq = fetch_receipt("IQ3_S", log).get(2) # the Coder's IQ1_M: the same LFS object, AUD-ARMS Part D) if iq and any(f"LINKED {quant} shard 2" in ln for ln in lines): out[2] = dict(iq, linked=True) return out def file_facts(quant, root=MODEL_ROOT, log=FETCH_LOG, rehash=False): """Per shard: the pinned sha, the fetch receipt's sha, size and mtime now, and (with rehash) a fresh sha.""" rec = fetch_receipt(quant, log) out = {} for i in (1, 2): p = shard_path(quant, i, root) st = os.stat(p) if os.path.exists(p) else None r = rec.get(i) or {} f = {"path": p, "pinned_sha256": SHA[quant][i], "receipt_sha256": r.get("sha256"), "receipt_at": r.get("at"), "bytes": st.st_size if st else None, "mtime": utc(st.st_mtime) if st else None, "receipt_bytes_match": (st is not None and r.get("bytes") == st.st_size)} if rehash and st: f["fresh_sha256"] = sha256_file(p) f["ok"] = bool(st) and r.get("sha256") == SHA[quant][i] and f["receipt_bytes_match"] and \ (not rehash or f.get("fresh_sha256") == SHA[quant][i]) out[i] = f return out