#!/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()