#!/usr/bin/env python3
"""Phase 0 + Phase 2 Local M5 Air Validation Runner (with Hard OOM / Swap Fuse)
=============================================================================
Measures cold prefill (PromptGemm), warm decode (tok/s), NVMe disk0 physical bytes
read (iostat -Id), and teacher-forced full-vocabulary (248,320) FP64 KLD / PPL /
top-1 agreement on K256_reap_l2_sum across llama.cpp and Strata Metal on the 13"
M5 MacBook Air (24 GB Unified Memory), while enforcing a 0.25s vm_stat watchdog
that immediately SIGKILLs the child process if clean available memory drops below
min_avail_gib (default 2.2 GiB) or swapouts increase by more than max_swapout_delta_mib
(default 64.0 MiB).
"""
import argparse
import ctypes
import datetime
import hashlib
import mmap
import json
import os
import re
import signal
import subprocess
import sys
import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional
sys.path.insert(0, str(Path(__file__).resolve().parent))
from slice_qwen4exp_gguf import parse_gguf_header
def _madvise_file(path: str, advice: int) -> bool:
libc = ctypes.CDLL("/usr/lib/libc.dylib", use_errno=True)
libc.mmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_longlong]
libc.mmap.restype = ctypes.c_void_p
libc.madvise.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_int]
libc.madvise.restype = ctypes.c_int
libc.munmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
libc.munmap.restype = ctypes.c_int
fd = os.open(path, os.O_RDONLY)
try:
sz = os.fstat(fd).st_size
if sz == 0:
return True
addr = libc.mmap(None, sz, 1, 1, fd, 0) # PROT_READ=1, MAP_SHARED=1
if addr is None or addr == ctypes.c_void_p(-1).value:
return False
try:
return libc.madvise(ctypes.c_void_p(addr), ctypes.c_size_t(sz), advice) == 0
finally:
libc.munmap(ctypes.c_void_p(addr), ctypes.c_size_t(sz))
finally:
os.close(fd)
def warm_file_cache(path: str) -> bool:
"""Advises MADV_WILLNEED (3) on path and faults pages so the first benchmark run on path does not cold-fault dense headers."""
return _madvise_file(path, 3)
def file_cache_residency(path: str) -> float:
"""Returns the fraction [0.0, 1.0] of pages of path currently resident in core via Darwin mincore(2)."""
libc = ctypes.CDLL("/usr/lib/libc.dylib", use_errno=True)
libc.mmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_longlong]
libc.mmap.restype = ctypes.c_void_p
libc.mincore.argtypes = [ctypes.c_void_p, ctypes.c_size_t, ctypes.c_char_p]
libc.mincore.restype = ctypes.c_int
libc.munmap.argtypes = [ctypes.c_void_p, ctypes.c_size_t]
libc.munmap.restype = ctypes.c_int
page_sz = os.sysconf("SC_PAGESIZE")
fd = os.open(path, os.O_RDONLY)
try:
sz = os.fstat(fd).st_size
if sz == 0:
return 1.0
n_pages = (sz + page_sz - 1) // page_sz
addr = libc.mmap(None, sz, 1, 1, fd, 0)
if addr is None or addr == ctypes.c_void_p(-1).value:
return 0.0
try:
vec = ctypes.create_string_buffer(n_pages)
if libc.mincore(ctypes.c_void_p(addr), ctypes.c_size_t(sz), vec) != 0:
return 0.0
in_core = sum(1 for b in vec.raw[:n_pages] if (b & 1) != 0)
return in_core / float(n_pages)
finally:
libc.munmap(ctypes.c_void_p(addr), ctypes.c_size_t(sz))
finally:
os.close(fd)
def sha256_file(path: Path) -> str:
h = hashlib.sha256()
with open(path, "rb") as f:
while True:
b = f.read(1 << 20)
if not b:
break
h.update(b)
return h.hexdigest()
def read_vm_stat() -> Dict[str, float]:
"""Parses macOS vm_stat and computes Strata's arena_avail metric."""
out = subprocess.check_output(["vm_stat"], text=True)
m_pg = re.search(r"page size of (\d+) bytes", out)
page_size = int(m_pg.group(1)) if m_pg else 16384
vals: Dict[str, int] = {}
for line in out.splitlines():
if ":" not in line:
continue
k, v = line.split(":", 1)
v_clean = v.strip().rstrip(".")
if v_clean.isdigit():
vals[k.strip()] = int(v_clean)
free = vals.get("Pages free", 0) * page_size
purgeable = vals.get("Pages purgeable", 0) * page_size
inactive = vals.get("Pages inactive", 0) * page_size
speculative = vals.get("Pages speculative", 0) * page_size
file_backed = vals.get("File-backed pages", 0) * page_size
wired = vals.get("Pages wired down", 0) * page_size
compressed = vals.get("Pages occupied by compressor", 0) * page_size
anon = vals.get("Anonymous pages", 0) * page_size
swapouts = vals.get("Swapouts", 0) * page_size
pageouts = vals.get("Pageouts", 0) * page_size
avail_bytes = free + purgeable + min(inactive + speculative, file_backed)
gib = float(1024 ** 3)
mib = float(1024 ** 2)
return {
"page_size": page_size,
"avail_gib": round(avail_bytes / gib, 3),
"free_gib": round(free / gib, 3),
"file_cache_gib": round(min(inactive + speculative, file_backed) / gib, 3),
"wired_gib": round(wired / gib, 3),
"compressor_gib": round(compressed / gib, 3),
"anon_gib": round(anon / gib, 3),
"swapouts_mib": round(swapouts / mib, 2),
"pageouts_mib": round(pageouts / mib, 2),
}
def read_iostat_disk0() -> Dict[str, float]:
"""Reads cumulative transfers and MB transferred on disk0 via iostat -Id."""
out = subprocess.check_output(["iostat", "-Id", "disk0"], text=True)
lines = [ln.strip() for ln in out.strip().splitlines() if ln.strip()]
parts = lines[-1].split()
return {
"kb_per_t": float(parts[0]),
"xfrs": float(parts[1]),
"mb": float(parts[2]),
}
class MemoryWatchdog:
"""Polls vm_stat every 0.25s while a subprocess is running and SIGKILLs on fuse trip."""
def __init__(self, proc: subprocess.Popen, min_avail_gib: float = 2.2, max_swapout_delta_mib: float = 64.0):
self.proc = proc
self.min_avail_gib = min_avail_gib
self.max_swapout_delta_mib = max_swapout_delta_mib
self.initial_vm = read_vm_stat()
self.min_seen_avail_gib = self.initial_vm["avail_gib"]
self.max_seen_wired_gib = self.initial_vm["wired_gib"]
self.max_seen_compressor_gib = self.initial_vm["compressor_gib"]
self.max_swapout_delta_seen_mib = 0.0
self.tripped_reason: Optional[str] = None
self._stop = threading.Event()
self._thread = threading.Thread(target=self._loop, daemon=True)
def start(self) -> None:
self._thread.start()
def stop(self) -> None:
self._stop.set()
self._thread.join(timeout=2.0)
def _loop(self) -> None:
while not self._stop.is_set():
if self.proc.poll() is not None:
break
try:
vm = read_vm_stat()
self.min_seen_avail_gib = min(self.min_seen_avail_gib, vm["avail_gib"])
self.max_seen_wired_gib = max(self.max_seen_wired_gib, vm["wired_gib"])
self.max_seen_compressor_gib = max(self.max_seen_compressor_gib, vm["compressor_gib"])
swap_delta = max(0.0, vm["swapouts_mib"] - self.initial_vm["swapouts_mib"])
self.max_swapout_delta_seen_mib = max(self.max_swapout_delta_seen_mib, swap_delta)
if vm["avail_gib"] < self.min_avail_gib:
self.tripped_reason = (
f"avail_gib dropped to {vm['avail_gib']:.2f} GiB (< {self.min_avail_gib:.2f} GiB fuse)"
)
os.kill(self.proc.pid, signal.SIGKILL)
break
if swap_delta > self.max_swapout_delta_mib:
self.tripped_reason = (
f"swapouts increased by {swap_delta:.1f} MiB (> {self.max_swapout_delta_mib:.1f} MiB cap)"
)
os.kill(self.proc.pid, signal.SIGKILL)
break
except Exception:
pass
time.sleep(0.25)
def parse_strata_output(text: str) -> Dict[str, Any]:
res: Dict[str, Any] = {}
m_res = re.search(
r"STRATA_RESULT init_ms=([\d.]+) prompt_tokens=(\d+) prefill_ms=([\d.]+) prefill_tok_s=([\d.]+) "
r"decode_tokens=(\d+) decode_ms=([\d.]+) decode_tok_s=([\d.]+) decode_ms_per_tok=([\d.]+)",
text,
)
if m_res:
res.update({
"init_ms": float(m_res.group(1)),
"prompt_tokens": int(m_res.group(2)),
"prefill_eval_ms": float(m_res.group(3)),
"prefill_tok_per_s": float(m_res.group(4)),
"eval_tokens": int(m_res.group(5)),
"eval_ms": float(m_res.group(6)),
"eval_tok_per_s": float(m_res.group(7)),
"eval_ms_per_tok": float(m_res.group(8)),
})
m_es = re.search(
r"expert stream\s+(\d+) windows: (\d+) distinct lookups \(([\d.]+) per window\), hits ([\d.]+), "
r"read-ahead hits ([\d.]+) \((\d+) of them late\), misses ([\d.]+) \(([\d.]+) per window\); "
r"read-aheads (\d+) \(([\d.]+) per window, (\d+) dropped\); ([\d.]+) MB read per window \((\d+) reads\); "
r"host ([\d.]+) ms per window, of it waiting ([\d.]+) ms",
text,
)
if m_es:
arena_hr = float(m_es.group(4))
ra_hr = float(m_es.group(5))
res.update({
"arena_hit_rate": arena_hr,
"readahead_hit_rate": ra_hr,
"combined_hit_rate": round(arena_hr + ra_hr, 4),
"late_readaheads": int(m_es.group(6)),
"miss_rate": float(m_es.group(7)),
"misses_per_tok": float(m_es.group(8)),
"mb_read_per_tok": float(m_es.group(12)),
"reads_total": int(m_es.group(13)),
"host_ms_per_tok": float(m_es.group(14)),
"host_wait_ms_per_tok": float(m_es.group(15)),
})
m_pr = re.search(
r"strata metal profile: window T1\s+(\d+) x\s+([\d.]+) ms: host before ([\d.]+), "
r"run_graph ([\d.]+) \(GPU ([\d.]+), so encode/submit/wait ([\d.]+)\), patch\+finish ([\d.]+)",
text,
)
if m_pr:
res["profile_window_t1_all_ms"] = float(m_pr.group(2))
res["profile_window_t1_run_graph_ms"] = float(m_pr.group(4))
res["profile_window_t1_gpu_incl_host_ms"] = float(m_pr.group(5))
if "host_ms_per_tok" in res:
res["pure_gpu_ms_per_tok"] = round(float(m_pr.group(5)) - res["host_ms_per_tok"], 2)
m_ar = re.search(
r"expert arena ([\d.]+) GB, (\d+) slots in (\d+) size classes, (\d+) experts read from the profile's top "
r"\(\d+ ranked\) in ([\d.]+) s \(([\d.]+) GB/s\)",
text,
)
if m_ar:
res.update({
"arena_gb": float(m_ar.group(1)),
"arena_slots": int(m_ar.group(2)),
"arena_size_classes": int(m_ar.group(3)),
"arena_fill_s": float(m_ar.group(5)),
"arena_fill_gb_per_s": float(m_ar.group(6)),
})
m_lk = re.search(
r"locked ([\d.]+) GB of active GPU dense tensors \(unpinned ([\d.]+) GB CPU token_embd \+ ([\d.]+) GB skipped BF16 HC in ([\d.]+) GB file map\)",
text,
)
if m_lk:
res.update({
"locked_dense_gb": float(m_lk.group(1)),
"unpinned_embd_gb": float(m_lk.group(2)),
"skipped_hc_gb": float(m_lk.group(3)),
"dense_file_map_gb": float(m_lk.group(4)),
})
m_un = re.search(
r"prompt scratch ([\d.]+) GB unmapped after prefill \(phys_footprint ([\d.]+) -> ([\d.]+) GB, resident RSS ([\d.]+) -> ([\d.]+) GB\)",
text,
)
if m_un:
res.update({
"prompt_scratch_unmapped_gb": float(m_un.group(1)),
"footprint_before_unmap_gb": float(m_un.group(2)),
"footprint_after_unmap_gb": float(m_un.group(3)),
"rss_before_unmap_gb": float(m_un.group(4)),
"rss_after_unmap_gb": float(m_un.group(5)),
})
m_kld = re.search(
r"STRATA_KLD eval_tokens=(\d+) ref_ppl=([\d.]+) cand_ppl=([\d.]+) delta_ppl=([-\d.]+) "
r"mean_kld=([\d.]+) median_kld=([\d.]+) p90_kld=([\d.]+) top1_agree_pct=([\d.]+)",
text,
)
if m_kld:
res.update({
"eval_tokens": int(m_kld.group(1)),
"ref_ppl": float(m_kld.group(2)),
"cand_ppl": float(m_kld.group(3)),
"delta_ppl": float(m_kld.group(4)),
"mean_kld": float(m_kld.group(5)),
"median_kld": float(m_kld.group(6)),
"p90_kld": float(m_kld.group(7)),
"top1_agree_pct": float(m_kld.group(8)),
})
m_kld2 = re.search(
r"STRATA_KLD_REF2 eval_tokens=(\d+) ref_ppl=([\d.]+) cand_ppl=([\d.]+) delta_ppl=([-\d.]+) "
r"mean_kld=([\d.]+) median_kld=([\d.]+) p90_kld=([\d.]+) top1_agree_pct=([\d.]+)",
text,
)
if m_kld2:
res["ref2"] = {
"eval_tokens": int(m_kld2.group(1)),
"ref_ppl": float(m_kld2.group(2)),
"cand_ppl": float(m_kld2.group(3)),
"delta_ppl": float(m_kld2.group(4)),
"mean_kld": float(m_kld2.group(5)),
"median_kld": float(m_kld2.group(6)),
"p90_kld": float(m_kld2.group(7)),
"top1_agree_pct": float(m_kld2.group(8)),
}
m_ppl = re.search(r"STRATA_PPL eval_tokens=(\d+) cand_ppl=([\d.]+)", text)
if m_ppl:
res.update({
"eval_tokens": int(m_ppl.group(1)),
"cand_ppl": float(m_ppl.group(2)),
})
return res
def run_guarded_llama(
cmd: List[str],
min_avail_gib: float = 2.2,
max_swapout_delta_mib: float = 64.0,
timeout_sec: int = 300,
env_overrides: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
"""Runs a benchmark subprocess under the 0.25s MemoryWatchdog + disk0 iostat delta."""
vm_before = read_vm_stat()
io_before = read_iostat_disk0()
t0 = time.monotonic()
run_env = os.environ.copy()
if env_overrides:
run_env.update(env_overrides)
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
errors="replace",
env=run_env,
)
wd = MemoryWatchdog(proc, min_avail_gib=min_avail_gib, max_swapout_delta_mib=max_swapout_delta_mib)
wd.start()
timed_out = False
try:
out, _ = proc.communicate(timeout=timeout_sec)
except subprocess.TimeoutExpired:
timed_out = True
os.kill(proc.pid, signal.SIGKILL)
out, _ = proc.communicate()
finally:
wd.stop()
dur = time.monotonic() - t0
vm_after = read_vm_stat()
io_after = read_iostat_disk0()
return {
"cmd": cmd,
"env": env_overrides or {},
"returncode": proc.returncode,
"timed_out": timed_out,
"oom_fuse_tripped": wd.tripped_reason is not None,
"oom_fuse_reason": wd.tripped_reason,
"duration_sec": round(dur, 2),
"min_avail_gib": round(wd.min_seen_avail_gib, 3),
"max_wired_gib": round(wd.max_seen_wired_gib, 3),
"max_compressor_gib": round(wd.max_seen_compressor_gib, 3),
"max_swapout_delta_mib": round(wd.max_swapout_delta_seen_mib, 2),
"disk0_delta_mb": round(max(0.0, io_after["mb"] - io_before["mb"]), 2),
"disk0_delta_xfrs": round(max(0.0, io_after["xfrs"] - io_before["xfrs"]), 1),
"vm_before": vm_before,
"vm_after": vm_after,
"perf": parse_strata_output(out),
"output_tail": out[-4500:],
}
def run_self_test() -> None:
vm = read_vm_stat()
assert vm["page_size"] >= 4096
assert vm["avail_gib"] > 0.5
io = read_iostat_disk0()
assert io["mb"] > 0.0
sample_strata = (
"strata metal stream: locked 2.90 GB of active GPU dense tensors (unpinned 0.34 GB CPU token_embd + 1.28 GB skipped BF16 HC in 4.53 GB file map)\n"
"strata metal stream: expert arena 3.40 GB, 2441 slots in 1 size classes, 2441 experts read from the profile's top (12288 ranked) in 0.5 s (6.80 GB/s)\n"
"strata metal stream: prompt scratch 1.36 GB unmapped after prefill (phys_footprint 4.83 -> 4.77 GB, resident RSS 7.50 -> 6.14 GB)\n"
"STRATA_RESULT init_ms=340.00 prompt_tokens=1024 prefill_ms=3200.00 prefill_tok_s=320.00 decode_tokens=63 decode_ms=4650.00 decode_tok_s=13.55 decode_ms_per_tok=73.81\n"
"STRATA_KLD eval_tokens=64 ref_ppl=14.0935 cand_ppl=14.4055 delta_ppl=0.3120 mean_kld=0.568263 median_kld=0.234097 p90_kld=1.783694 top1_agree_pct=71.88\n"
"expert stream 63 windows: 30240 distinct lookups (480.0 per window), hits 0.8566, read-ahead hits 0.0566 (114 of them late), misses 0.0868 (41.7 per window); read-aheads 2263 (35.9 per window, 0 dropped); 99.9 MB read per window (4913 reads); host 16.70 ms per window, of it waiting 16.05 ms\n"
)
p = parse_strata_output(sample_strata)
assert p["prefill_tok_per_s"] == 320.0
assert p["combined_hit_rate"] == 0.9132
assert p["mean_kld"] == 0.568263
assert p["locked_dense_gb"] == 2.90
assert p["prompt_scratch_unmapped_gb"] == 1.36
assert p["rss_before_unmap_gb"] == 7.50 and p["rss_after_unmap_gb"] == 6.14
import fcntl, tempfile
with tempfile.NamedTemporaryFile(delete=False) as tf:
tmp_p = tf.name
try:
fd = os.open(tmp_p, os.O_WRONLY | os.O_TRUNC)
try:
fcntl.fcntl(fd, 48, 1) # F_NOCACHE = 48 on Darwin
os.write(fd, b"\x55" * (16 * 16384))
os.fsync(fd)
finally:
os.close(fd)
res_before = file_cache_residency(tmp_p)
assert res_before == 0.0, f"Expected 0.0 residency after F_NOCACHE write, got {res_before}"
assert warm_file_cache(tmp_p) is True
res_after = file_cache_residency(tmp_p)
assert res_after == 1.0, f"Expected 1.0 residency after warm_file_cache, got {res_after}"
finally:
if os.path.exists(tmp_p):
os.unlink(tmp_p)
print("scripts/run_phase0_phase2_m5_validation.py self-test: PASS")
def main() -> None:
parser = argparse.ArgumentParser(description="Phase 0 + Phase 2 Local M5 Air Validation Runner")
parser.add_argument("--self-test", action="store_true", help="Run unit self-tests and exit")
parser.add_argument(
"--strata-cli",
default="vendor/strata/build-metal/strata_window_cli",
help="Path to strata_window_cli binary",
)
parser.add_argument(
"--ple-gguf",
default="models/K256_reap_l2_sum/Swift-Qwen3.8-Flash-Next-GSQ-RCO-IQ3_S-K256_reap_l2_sum-00001-of-00002.gguf",
help="Path to Shard 1 GGUF containing PLE table",
)
parser.add_argument("--min-avail-gib", type=float, default=2.2, help="Hard OOM fuse floor in GiB")
parser.add_argument("--max-swapout-delta-mib", type=float, default=64.0, help="Hard swapout delta cap in MiB")
parser.add_argument(
"--output",
default="results/phase0_phase2_m5_validation.json",
help="Output JSON path",
)
args = parser.parse_args()
if args.self_test:
run_self_test()
return
strata_cli = Path(args.strata_cli).resolve()
ple_gguf = str(Path(args.ple_gguf).resolve())
iq3s_gguf = str(Path("models/K256_reap_l2_sum/K256-metal.gguf").resolve())
q2_0_gguf = str(Path("models/K256_reap_l2_sum/K256-q2_0-metal.gguf").resolve())
prof_bin = str(Path("models/K256_reap_l2_sum/K256-expert-profile.bin").resolve())
base_env = {
"STRATA_METAL_EXPERT_PROFILE": prof_bin,
"STRATA_METAL_ARENA_GB": "3.6",
"STRATA_METAL_WIRE_ARENA": "0",
"STRATA_METAL_MLOCK_DENSE": "1",
"STRATA_METAL_SEPARATE_PROMPT_BUF": "1",
"STRATA_METAL_PROMPT_GEMM": "1",
"STRATA_METAL_PROMPT_GEMM_MIN": "16",
"STRATA_METAL_PROMPT_CHUNK": "1024",
"STRATA_METAL_READ_SPLIT": "2",
"STRATA_METAL_IO_THREADS": "8",
"STRATA_METAL_PREFETCH": "8",
"STRATA_METAL_PREFETCH2": "0",
"STRATA_METAL_PREFETCH_MRU": "0",
"STRATA_METAL_MATH": "fast",
"STRATA_METAL_PROFILE": "1",
}
# Load archived llama.cpp baseline runs if present
prev_llama_unpatched = []
prev_llama_patched = []
out_path = Path(args.output)
if out_path.exists():
old_data = json.loads(out_path.read_text(encoding="utf-8"))
prev_llama_unpatched = old_data.get("llama_cpp_unpatched_runs", [])
prev_llama_patched = old_data.get("llama_cpp_patched_runs", [])
unpatched_path = Path("results/phase0_phase2_m5_unpatched_baseline.json")
if not prev_llama_unpatched and unpatched_path.exists():
prev_llama_unpatched = json.loads(unpatched_path.read_text(encoding="utf-8")).get("runs", [])
warm_file_cache(iq3s_gguf)
strata_speed_runs: List[Dict[str, Any]] = []
strata_kld_runs: List[Dict[str, Any]] = []
# Phase A: Run all 5 speed benchmark runs first while the fanless M5 Air is thermally unthrottled
print("=== Running strata_iq3s_moce0_short_43tok (MOCE=0.0, n_prompt=43) ===", flush=True)
env_iq3 = dict(base_env)
env_iq3["STRATA_METAL_MOCE_BONUS"] = "0.0"
env_iq3["STRATA_METAL_ARENA_GB"] = "3.4"
r_iq3_speed = run_guarded_llama(
[
str(strata_cli),
"--metal-gguf", iq3s_gguf,
"--ple-gguf", ple_gguf,
"--tokens-file", str(Path("data/m5_val_prompts/prompt_43tok_wal.txt").resolve()),
"--n-expert", "256",
"--ctx", "4096",
"--n-prompt", "43",
"--n-gen", "64",
],
min_avail_gib=args.min_avail_gib,
max_swapout_delta_mib=args.max_swapout_delta_mib,
timeout_sec=180,
env_overrides=env_iq3,
)
r_iq3_speed["step"] = "strata_iq3s_moce0_short_43tok"
strata_speed_runs.append(r_iq3_speed)
p = r_iq3_speed["perf"]
print(
f" exit={r_iq3_speed['returncode']} swap={r_iq3_speed['max_swapout_delta_mib']} MiB "
f"prefill={p.get('prefill_tok_per_s')} tok/s decode={p.get('eval_tok_per_s')} tok/s "
f"hit={p.get('combined_hit_rate')} misses/tok={p.get('misses_per_tok')} mb/tok={p.get('mb_read_per_tok')}",
flush=True,
)
warm_file_cache(q2_0_gguf)
# Phase B: Run all Q2_0 speed runs at uniform 3.40 GB arena (MOCE=0.0 and MOCE=0.85 across 43 / 1024 / 2048 tokens)
q2_speed_specs = [
("strata_q2_0_moce0_short_43tok", q2_0_gguf, "0.0", "data/m5_val_prompts/prompt_43tok_wal.txt", 43, 64, "3.4"),
("strata_q2_0_moce085_short_43tok", q2_0_gguf, "0.85", "data/m5_val_prompts/prompt_43tok_wal.txt", 43, 64, "3.4"),
("strata_q2_0_moce085_long_1024tok", q2_0_gguf, "0.85", "data/m5_val_prompts/prompt_1024tok_ops.txt", 1024, 64, "3.4"),
("strata_q2_0_moce085_long_2048tok", q2_0_gguf, "0.85", "data/m5_val_prompts/prompt_2048tok_ops.txt", 2048, 64, "3.4"),
]
for step_name, gguf_path, moce_val, pfile, n_p, n_g, arena_gb in q2_speed_specs:
print(f"=== Running {step_name} (MOCE={moce_val}, n_prompt={n_p}) ===", flush=True)
env = dict(base_env)
env["STRATA_METAL_MOCE_BONUS"] = moce_val
env["STRATA_METAL_ARENA_GB"] = arena_gb
cmd = [
str(strata_cli),
"--metal-gguf", gguf_path,
"--ple-gguf", ple_gguf,
"--tokens-file", str(Path(pfile).resolve()),
"--n-expert", "256",
"--ctx", "4096",
"--n-prompt", str(n_p),
"--n-gen", str(n_g),
]
r = run_guarded_llama(
cmd,
min_avail_gib=args.min_avail_gib,
max_swapout_delta_mib=args.max_swapout_delta_mib,
timeout_sec=180,
env_overrides=env,
)
r["step"] = step_name
strata_speed_runs.append(r)
p = r["perf"]
print(
f" exit={r['returncode']} swap={r['max_swapout_delta_mib']} MiB "
f"prefill={p.get('prefill_tok_per_s')} tok/s decode={p.get('eval_tok_per_s')} tok/s "
f"hit={p.get('combined_hit_rate')} misses/tok={p.get('misses_per_tok')} mb/tok={p.get('mb_read_per_tok')}",
flush=True,
)
# Phase C: Run 3-domain teacher-forced full-vocab (248,320) FP64 KLD, PPL & Top-1 Agreement across ops_tools, chat_json, ja
warm_file_cache(iq3s_gguf)
for dom in ("ops_tools", "chat_json", "ja"):
step_name = f"tf_{dom}_iq3s_moce0_ref"
pfile_path = Path(f"data/m5_val_prompts/eval_tf_{dom}.txt").resolve()
n_total_toks = len(pfile_path.read_text(encoding="utf-8").split())
n_tf_gen = 64
n_tf_prompt = n_total_toks - n_tf_gen
ref_iq3_bin = f"/tmp/strata_ref_iq3s_{dom}.bin"
print(f"=== Running {step_name} (n_prompt={n_tf_prompt}, n_gen={n_tf_gen}) ===", flush=True)
r = run_guarded_llama(
[
str(strata_cli),
"--metal-gguf", iq3s_gguf,
"--ple-gguf", ple_gguf,
"--tokens-file", str(pfile_path),
"--n-expert", "256",
"--ctx", "4096",
"--n-prompt", str(n_tf_prompt),
"--n-gen", str(n_tf_gen),
"--logits-out", ref_iq3_bin,
],
min_avail_gib=args.min_avail_gib,
max_swapout_delta_mib=args.max_swapout_delta_mib,
timeout_sec=180,
env_overrides=env_iq3,
)
r["step"] = step_name
r["domain"] = dom
strata_kld_runs.append(r)
p = r["perf"]
print(f" exit={r['returncode']} swap={r['max_swapout_delta_mib']} MiB cand_ppl={p.get('cand_ppl')}", flush=True)
warm_file_cache(q2_0_gguf)
for dom in ("ops_tools", "chat_json", "ja"):
pfile_path = Path(f"data/m5_val_prompts/eval_tf_{dom}.txt").resolve()
n_total_toks = len(pfile_path.read_text(encoding="utf-8").split())
n_tf_gen = 64
n_tf_prompt = n_total_toks - n_tf_gen
pfile = str(pfile_path)
ref_iq3_bin = f"/tmp/strata_ref_iq3s_{dom}.bin"
ref_q2_m0_bin = f"/tmp/strata_ref_q2_m0_{dom}.bin"
# 1. Q2_0 (MOCE=0.0) vs IQ3_S (MOCE=0.0)
step_m0 = f"tf_{dom}_q2_0_moce0_vs_iq3s"
print(f"=== Running {step_m0} (n_prompt={n_tf_prompt}, n_gen={n_tf_gen}) ===", flush=True)
env_m0 = dict(base_env)
env_m0["STRATA_METAL_MOCE_BONUS"] = "0.0"
env_m0["STRATA_METAL_ARENA_GB"] = "3.4"
r_m0 = run_guarded_llama(
[
str(strata_cli),
"--metal-gguf", q2_0_gguf,
"--ple-gguf", ple_gguf,
"--tokens-file", pfile,
"--n-expert", "256",
"--ctx", "4096",
"--n-prompt", str(n_tf_prompt),
"--n-gen", str(n_tf_gen),
"--logits-out", ref_q2_m0_bin,
"--logits-ref", ref_iq3_bin,
],
min_avail_gib=args.min_avail_gib,
max_swapout_delta_mib=args.max_swapout_delta_mib,
timeout_sec=180,
env_overrides=env_m0,
)
r_m0["step"] = step_m0
r_m0["domain"] = dom
strata_kld_runs.append(r_m0)
p0 = r_m0["perf"]
print(
f" exit={r_m0['returncode']} swap={r_m0['max_swapout_delta_mib']} MiB "
f"cand_ppl={p0.get('cand_ppl')} mean_kld={p0.get('mean_kld')} top1={p0.get('top1_agree_pct')}%",
flush=True,
)
# 2. Q2_0 (MOCE=0.85) vs IQ3_S (MOCE=0.0) [STRATA_KLD] AND vs Q2_0 (MOCE=0.0) [STRATA_KLD_REF2] in a single pass
step_m85 = f"tf_{dom}_q2_0_moce085_vs_iq3s"
print(f"=== Running {step_m85} (+ ref2 vs q2_0_moce0, n_prompt={n_tf_prompt}, n_gen={n_tf_gen}) ===", flush=True)
env_m85 = dict(base_env)
env_m85["STRATA_METAL_MOCE_BONUS"] = "0.85"
env_m85["STRATA_METAL_ARENA_GB"] = "3.4"
r_m85 = run_guarded_llama(
[
str(strata_cli),
"--metal-gguf", q2_0_gguf,
"--ple-gguf", ple_gguf,
"--tokens-file", pfile,
"--n-expert", "256",
"--ctx", "4096",
"--n-prompt", str(n_tf_prompt),
"--n-gen", str(n_tf_gen),
"--logits-ref", ref_iq3_bin,
"--logits-ref2", ref_q2_m0_bin,
],
min_avail_gib=args.min_avail_gib,
max_swapout_delta_mib=args.max_swapout_delta_mib,
timeout_sec=180,
env_overrides=env_m85,
)
r_m85["step"] = step_m85
r_m85["domain"] = dom
strata_kld_runs.append(r_m85)
p85 = r_m85["perf"]
p85_ref2 = p85.get("ref2", {})
print(
f" exit={r_m85['returncode']} swap={r_m85['max_swapout_delta_mib']} MiB "
f"cand_ppl={p85.get('cand_ppl')} vs_iq3s_kld={p85.get('mean_kld')} ({p85.get('top1_agree_pct')}%) "
f"vs_q2_m0_kld={p85_ref2.get('mean_kld')} ({p85_ref2.get('top1_agree_pct')}%)",
flush=True,
)
for tmp_f in (ref_iq3_bin, ref_q2_m0_bin):
if Path(tmp_f).exists():
Path(tmp_f).unlink()
# Build summary_comparison strictly from the archived runs
by_step = {r["step"]: r for r in strata_speed_runs}
s_iq3 = by_step["strata_iq3s_moce0_short_43tok"]["perf"]
s_q2_m0 = by_step["strata_q2_0_moce0_short_43tok"]["perf"]
s_q2_m85 = by_step["strata_q2_0_moce085_short_43tok"]["perf"]
s_1k = by_step["strata_q2_0_moce085_long_1024tok"]["perf"]
s_2k = by_step["strata_q2_0_moce085_long_2048tok"]["perf"]
kld_by_step = {r["step"]: r["perf"] for r in strata_kld_runs}
q2_vs_iq3_klds = [kld_by_step[f"tf_{d}_q2_0_moce0_vs_iq3s"]["mean_kld"] for d in ("ops_tools", "chat_json", "ja")]
q2_vs_iq3_top1 = [kld_by_step[f"tf_{d}_q2_0_moce0_vs_iq3s"]["top1_agree_pct"] for d in ("ops_tools", "chat_json", "ja")]
moce_vs_q2_klds = [kld_by_step[f"tf_{d}_q2_0_moce085_vs_iq3s"]["ref2"]["mean_kld"] for d in ("ops_tools", "chat_json", "ja")]
moce_vs_q2_top1 = [kld_by_step[f"tf_{d}_q2_0_moce085_vs_iq3s"]["ref2"]["top1_agree_pct"] for d in ("ops_tools", "chat_json", "ja")]
comb_vs_iq3_klds = [kld_by_step[f"tf_{d}_q2_0_moce085_vs_iq3s"]["mean_kld"] for d in ("ops_tools", "chat_json", "ja")]
comb_vs_iq3_top1 = [kld_by_step[f"tf_{d}_q2_0_moce085_vs_iq3s"]["top1_agree_pct"] for d in ("ops_tools", "chat_json", "ja")]
miss_red_vs_q2_m0_pct = round(100.0 * (s_q2_m0["misses_per_tok"] - s_q2_m85["misses_per_tok"]) / s_q2_m0["misses_per_tok"], 1)
miss_red_vs_iq3_m0_pct = round(100.0 * (s_iq3["misses_per_tok"] - s_q2_m85["misses_per_tok"]) / s_iq3["misses_per_tok"], 1)
lu_by_step = {r["step"]: r for r in prev_llama_unpatched}
lp_by_step = {r["step"]: r for r in prev_llama_patched}
lu_1024 = lu_by_step["step2_ngl99_cmoe_ub1024"]
lu_2048 = lu_by_step["step2_ngl99_cmoe_ub2048"]
lp_s1 = lp_by_step["step1_ngl99_cmoe_short_prompt"]
lp_1024 = lp_by_step["step2_ngl99_cmoe_ub1024"]
lp_2048 = lp_by_step["step2_ngl99_cmoe_ub2048"]
lp_mc = lp_by_step["step3_ngl99_cmoe_moecache1536"]
summary = {
"llama_cpp_unpatched_cmoe": {
"prefill_1264tok_ub1024_tok_per_s": lu_1024["perf"]["prompt_tok_per_s"],
"prefill_1264tok_ub2048_tok_per_s": lu_2048["perf"]["prompt_tok_per_s"],
"decode_tok_per_s": lu_1024["perf"]["eval_tok_per_s"],
"whole_run_disk0_mb_1264tok_ub1024": lu_1024["disk0_delta_mb"],
"whole_run_disk0_mb_1264tok_ub2048": lu_2048["disk0_delta_mb"],
"swapouts_mib": max(r["max_swapout_delta_mib"] for r in prev_llama_unpatched),
},
"llama_cpp_patched_m5_lazy_moe": {
"prefill_1264tok_ub1024_tok_per_s": lp_1024["perf"]["prompt_tok_per_s"],
"prefill_1264tok_ub2048_tok_per_s": lp_2048["perf"]["prompt_tok_per_s"],
"decode_22tok_prompt_tok_per_s": lp_s1["perf"]["eval_tok_per_s"],
"decode_moe_cache_1536mib_tok_per_s": lp_mc["perf"]["eval_tok_per_s"],
"whole_run_disk0_mb_22tok": lp_s1["disk0_delta_mb"],
"whole_run_disk0_mb_1264tok_ub1024_prefill_only": lp_1024["disk0_delta_mb"],
"whole_run_disk0_mb_1264tok_ub2048_prefill_only": lp_2048["disk0_delta_mb"],
"swapouts_mib": max(r["max_swapout_delta_mib"] for r in prev_llama_patched),
},
"strata_metal_iq3s_moce0_43tok": {
"init_ms": s_iq3["init_ms"],
"decode_tok_per_s": s_iq3["eval_tok_per_s"],
"decode_ms_per_tok": s_iq3["eval_ms_per_tok"],
"arena_hit_rate": s_iq3["arena_hit_rate"],
"readahead_hit_rate": s_iq3["readahead_hit_rate"],
"combined_hit_rate": s_iq3["combined_hit_rate"],
"misses_per_tok": s_iq3["misses_per_tok"],
"ssd_read_mb_per_decode_tok": s_iq3["mb_read_per_tok"],
},
"strata_metal_q2_0_moce0_43tok": {
"init_ms": s_q2_m0["init_ms"],
"decode_tok_per_s": s_q2_m0["eval_tok_per_s"],
"decode_ms_per_tok": s_q2_m0["eval_ms_per_tok"],
"arena_hit_rate": s_q2_m0["arena_hit_rate"],
"readahead_hit_rate": s_q2_m0["readahead_hit_rate"],
"combined_hit_rate": s_q2_m0["combined_hit_rate"],
"misses_per_tok": s_q2_m0["misses_per_tok"],
"ssd_read_mb_per_decode_tok": s_q2_m0["mb_read_per_tok"],
},
"strata_metal_optimized_k256_q2_0_moce085": {
"init_ms_short_43tok": s_q2_m85["init_ms"],
"init_ms_1024tok": s_1k["init_ms"],
"init_ms_2048tok": s_2k["init_ms"],
"prefill_1024tok_eval_ms": s_1k["prefill_eval_ms"],
"prefill_2048tok_eval_ms": s_2k["prefill_eval_ms"],
"prefill_1024tok_tok_per_s": s_1k["prefill_tok_per_s"],
"prefill_2048tok_tok_per_s": s_2k["prefill_tok_per_s"],
"decode_43tok_tok_per_s": s_q2_m85["eval_tok_per_s"],
"decode_1024tok_tok_per_s": s_1k["eval_tok_per_s"],
"decode_2048tok_tok_per_s": s_2k["eval_tok_per_s"],
"decode_43tok_ms_per_tok": s_q2_m85["eval_ms_per_tok"],
"decode_1024tok_ms_per_tok": s_1k["eval_ms_per_tok"],
"decode_2048tok_ms_per_tok": s_2k["eval_ms_per_tok"],
"arena_hit_rate_43tok": s_q2_m85["arena_hit_rate"],
"readahead_hit_rate_43tok": s_q2_m85["readahead_hit_rate"],
"combined_hit_rate_43tok": s_q2_m85["combined_hit_rate"],
"arena_hit_rate_1024tok": s_1k["arena_hit_rate"],
"readahead_hit_rate_1024tok": s_1k["readahead_hit_rate"],
"combined_hit_rate_1024tok": s_1k["combined_hit_rate"],
"arena_hit_rate_2048tok": s_2k["arena_hit_rate"],
"readahead_hit_rate_2048tok": s_2k["readahead_hit_rate"],
"combined_hit_rate_2048tok": s_2k["combined_hit_rate"],
"host_wait_ms_per_tok_43tok": s_q2_m85["host_wait_ms_per_tok"],
"pure_gpu_ms_per_tok_43tok": s_q2_m85["pure_gpu_ms_per_tok"],
"host_wait_ms_per_tok_1024tok": s_1k["host_wait_ms_per_tok"],
"pure_gpu_ms_per_tok_1024tok": s_1k["pure_gpu_ms_per_tok"],
"host_wait_ms_per_tok_2048tok": s_2k["host_wait_ms_per_tok"],
"pure_gpu_ms_per_tok_2048tok": s_2k["pure_gpu_ms_per_tok"],
"arena_fill_gb_per_s_43tok": s_q2_m85["arena_fill_gb_per_s"],
"arena_fill_gb_per_s_1024tok": s_1k["arena_fill_gb_per_s"],
"arena_fill_gb_per_s_2048tok": s_2k["arena_fill_gb_per_s"],
"misses_per_tok_43tok": s_q2_m85["misses_per_tok"],
"misses_per_tok_1024tok": s_1k["misses_per_tok"],
"misses_per_tok_2048tok": s_2k["misses_per_tok"],
"miss_reduction_vs_q2_0_moce0_pct": miss_red_vs_q2_m0_pct,
"miss_reduction_vs_iq3s_moce0_pct": miss_red_vs_iq3_m0_pct,
"ssd_read_mb_per_decode_tok_43tok": s_q2_m85["mb_read_per_tok"],
"ssd_read_mb_per_decode_tok_1024tok": s_1k["mb_read_per_tok"],
"ssd_read_mb_per_decode_tok_2048tok": s_2k["mb_read_per_tok"],
"arena_gb": s_1k["arena_gb"],
"arena_slots": s_1k["arena_slots"],
"locked_dense_gb": s_1k["locked_dense_gb"],
"unpinned_embd_gb": s_1k["unpinned_embd_gb"],
"skipped_hc_gb": s_1k["skipped_hc_gb"],
"prompt_scratch_unmapped_gb": s_1k["prompt_scratch_unmapped_gb"],
"footprint_before_unmap_gb_1024tok": s_1k["footprint_before_unmap_gb"],
"footprint_after_unmap_gb_1024tok": s_1k["footprint_after_unmap_gb"],
"rss_before_unmap_gb_1024tok": s_1k["rss_before_unmap_gb"],
"rss_after_unmap_gb_1024tok": s_1k["rss_after_unmap_gb"],
"swapouts_mib": max(r["max_swapout_delta_mib"] for r in strata_speed_runs),
},
"quality_kld_3domain_teacher_forced_192tok": {
"q2_0_moce0_vs_iq3s_moce0": {
"ops_tools": kld_by_step["tf_ops_tools_q2_0_moce0_vs_iq3s"],
"chat_json": kld_by_step["tf_chat_json_q2_0_moce0_vs_iq3s"],
"ja": kld_by_step["tf_ja_q2_0_moce0_vs_iq3s"],
"macro_mean_kld": round(sum(q2_vs_iq3_klds) / 3.0, 6),
"macro_top1_agree_pct": round(sum(q2_vs_iq3_top1) / 3.0, 2),
},
"q2_0_moce085_vs_q2_0_moce0": {
"ops_tools": kld_by_step["tf_ops_tools_q2_0_moce085_vs_iq3s"]["ref2"],
"chat_json": kld_by_step["tf_chat_json_q2_0_moce085_vs_iq3s"]["ref2"],
"ja": kld_by_step["tf_ja_q2_0_moce085_vs_iq3s"]["ref2"],
"macro_mean_kld": round(sum(moce_vs_q2_klds) / 3.0, 6),
"macro_top1_agree_pct": round(sum(moce_vs_q2_top1) / 3.0, 2),
},
"q2_0_moce085_vs_iq3s_moce0": {
"ops_tools": kld_by_step["tf_ops_tools_q2_0_moce085_vs_iq3s"],
"chat_json": kld_by_step["tf_chat_json_q2_0_moce085_vs_iq3s"],
"ja": kld_by_step["tf_ja_q2_0_moce085_vs_iq3s"],
"macro_mean_kld": round(sum(comb_vs_iq3_klds) / 3.0, 6),
"macro_top1_agree_pct": round(sum(comb_vs_iq3_top1) / 3.0, 2),
},
},
}
provenance = {
"generated_at_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
"run_script_sha256": sha256_file(Path(__file__).resolve()),
"patch_strata_metal_sha256": sha256_file(Path("scripts/patch_strata_metal.py").resolve()),
"patch_m5_lazy_moe_sha256": sha256_file(Path("scripts/patch_m5_lazy_moe.py").resolve()),
"prepare_strata_k256_sha256": sha256_file(Path("scripts/prepare_strata_k256.py").resolve()),
"strata_window_cli_sha256": sha256_file(strata_cli),
"prompt_files_sha256": {
p.name: sha256_file(p) for p in sorted(Path("data/m5_val_prompts").glob("*.txt"))
},
}
report = {
"hardware": "13-inch MacBook Air (Mac17,3, Apple M5 10-core GPU, 24 GB Unified Memory, 1 TB NVMe)",
"provenance": provenance,
"watchdog_policy": {
"poll_interval_sec": 0.25,
"min_avail_gib": args.min_avail_gib,
"max_swapout_delta_mib": args.max_swapout_delta_mib,
},
"summary_comparison": summary,
"strata_optimized_runs": strata_speed_runs,
"strata_quality_kld_runs": strata_kld_runs,
"llama_cpp_patched_runs": prev_llama_patched,
"llama_cpp_unpatched_runs": prev_llama_unpatched,
}
out_path.parent.mkdir(parents=True, exist_ok=True)
out_path.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
print(f"\nSaved complete M5 validation report -> {out_path}", flush=True)
if __name__ == "__main__":
main()