#!/usr/bin/env python3 """AM-5 (clean build time) and AM-9 (peak RSS) — measured for the first time. CB-WP-0006 T03. Both rows were `unmutatable`: AM-5 is declared `recorded not gated` and had never been recorded, and AM-9 was declared "very unlikely to bind", which `evidence/CB-EV-0001` itself flags as an unmeasured judgment call. The task said: measure or withdraw, but stop leaving them blank. Measured. They came out differently, and one of them is a breach. **AM-5 does not build into `target/`.** It builds into a temporary `CARGO_TARGET_DIR` so a clean-build measurement does not destroy the working cache — otherwise measuring the metric would cost several minutes of rebuild every time, and a metric that punishes its own measurement gets measured once and never again. **AM-5 exits 0 even on a breach**, because `specs/GameKernel.md` §5 declares it `recorded not gated`. Gating it is a spec change and needs an ADR; this tool reports, loudly, and does not quietly promote itself. AM-9 carries no such declaration and **is** gated. Usage: python3 tools/runtime-metrics.py # both (AM-5 takes ~90 s) python3 tools/runtime-metrics.py --fast # AM-9 only, for `make all` python3 tools/runtime-metrics.py --self-test """ import glob import os import re import resource import shutil import subprocess import sys import tempfile import time from repo import ROOT, cargo_bin, cargo_env, enter_root AM5_MAX_SECONDS = 60.0 AM5_MACHINE = "bnt-lap001" # the spec names the machine; so do we # CB-WP-0006 T04 correction. The first AM-5 measurement read 87.0 s and was # reported as a 45% breach. Quiet re-runs read 54.1 s and 57.6 s — the row # passes. The 87 s was taken while the machine was busy with mutation-check # and cargo builds: a timing measurement under contention measures the # contention, which is the same error AM-6 had two tasks earlier. # # So AM-5 now (a) refuses to measure on a loaded machine and (b) takes the # BEST of N. Best, not worst: a build-time *ceiling* asks "can this machine # do it in 60 s", so the least-contended sample is the honest one — the # mirror of AM-6's best-of-N for a throughput *floor*. AM5_SAMPLES = 3 AM5_MAX_LOAD_PER_CPU = 0.5 # refuse above this; the machine is busy AM5_SPREAD_WARN = 1.25 # max/min above this = not a quiet run AM9_MAX_RSS_MB = 64.0 AM9_TEST = "replay_probe::replay_100k_events_is_linear_and_fast" def load_per_cpu(): """1-minute load average per CPU. > ~0.5 means real competing work.""" try: return os.getloadavg()[0] / (os.cpu_count() or 1) except OSError: return 0.0 def clean_build_seconds(extra=()): """Wall seconds for a clean release build, into a throwaway target dir.""" cargo = cargo_bin() if not cargo: return None tmp = tempfile.mkdtemp(prefix="cb-am5-") env = cargo_env() env["CARGO_TARGET_DIR"] = tmp try: t = time.monotonic() r = subprocess.run([cargo, "build", "--release", "--workspace", *extra], cwd=ROOT, env=env, capture_output=True, text=True) secs = time.monotonic() - t if r.returncode != 0: return None return secs finally: shutil.rmtree(tmp, ignore_errors=True) def test_binary(): """Newest release test binary for games-ground, built if absent.""" cargo = cargo_bin() subprocess.run([cargo, "test", "--release", "-p", "games-ground", "--all-features", "--no-run"], cwd=ROOT, env=cargo_env(), capture_output=True, text=True) cands = [p for p in glob.glob(os.path.join(ROOT, "target/release/deps/games_ground-*")) if not p.endswith(".d") and os.access(p, os.X_OK)] return max(cands, key=os.path.getmtime) if cands else None def peak_rss_mb(binary, test=AM9_TEST): """(peak RSS in MB, ok) for one test run, from that child's rusage. Uses `os.wait4`, which returns the rusage of **that specific child**. The obvious alternative — `getrusage(RUSAGE_CHILDREN)` — is a high-water mark across *every* reaped child of this process, so it attributed `cargo`'s memory to the workload and reported **38.2 MB for a run that actually used 12.3 MB**. Cross-checked against `/usr/bin/time -v` on the same binary, which is how the discrepancy was found. """ out_r, out_w = os.pipe() try: pid = os.posix_spawn( binary, [binary, test, "--exact"], os.environ, file_actions=[(os.POSIX_SPAWN_DUP2, out_w, 1), (os.POSIX_SPAWN_DUP2, out_w, 2)]) os.close(out_w) out_w = None chunks = [] while True: b = os.read(out_r, 65536) if not b: break chunks.append(b) _, _status, ru = os.wait4(pid, 0) finally: if out_w is not None: os.close(out_w) os.close(out_r) text = b"".join(chunks).decode(errors="replace") # ru_maxrss is kilobytes on Linux. return ru.ru_maxrss / 1024.0, "1 passed" in text def report(fast=False): rc = 0 print("AM-9 / M-D3-MEM — peak resident memory, 100k-event run") binary = test_binary() if not binary: print(" ERROR — could not locate the release test binary", file=sys.stderr) return 1 rss, ran = peak_rss_mb(binary) # Positive control: a run that did not execute the workload must not # be scored as low memory. Reporting 2 MB because the filter matched # nothing is exactly the harness-does-nothing shape. if not ran: print(f" ERROR — {AM9_TEST} did not run; refusing to report RSS", file=sys.stderr) return 1 ok9 = rss <= AM9_MAX_RSS_MB print(f" {rss:.1f} MB peak RSS [{'ok ' if ok9 else 'FAIL'} " f"target <= {AM9_MAX_RSS_MB:.0f} MB] " f"({AM9_MAX_RSS_MB / rss:.1f}x headroom)") if not ok9: rc = 2 if fast: print("\nAM-5 skipped (--fast). Run `make build-time` for the clean " "build measurement.") return rc print(f"\nAM-5 / M-D2-BLD — clean release build on {AM5_MACHINE}") host = subprocess.run(["hostname"], capture_output=True, text=True).stdout.strip() if host != AM5_MACHINE: print(f" NOTE: running on {host!r}, not {AM5_MACHINE!r} — the spec " f"target is machine-specific, so this is directional only.") # Positive control: refuse to measure on a busy machine rather than # publish the contention as a build time. This is the guard T03 did # not have, and its absence produced a reported 45% breach that was # not real. load = load_per_cpu() if load > AM5_MAX_LOAD_PER_CPU: print(f" ABORT — load average is {load:.2f} per CPU (limit " f"{AM5_MAX_LOAD_PER_CPU}); a build timed under contention " f"measures the contention. Re-run on a quiet machine.", file=sys.stderr) return 1 print(f" load before measuring: {load:.2f} per CPU over " f"{os.cpu_count()} CPUs — quiet") for label, extra in (("dev toolchain (default features)", ()), ("shipped runtime (--no-default-features)", ("--no-default-features",))): samples = [] for _ in range(AM5_SAMPLES): secs = clean_build_seconds(extra) if secs is None: print(f" ERROR — clean build failed: {label}", file=sys.stderr) return 1 samples.append(secs) best, worst = min(samples), max(samples) mark = "ok " if best <= AM5_MAX_SECONDS else "FAIL" spread = worst / best if best else 1.0 print(f" {label:<42} {best:6.1f} s [{mark} target " f"<= {AM5_MAX_SECONDS:.0f} s]") print(f" best of {AM5_SAMPLES}: " + ", ".join(f"{x:.1f}" for x in sorted(samples)) + f" (spread {spread:.2f}x)") if spread > AM5_SPREAD_WARN: print(f" WARN — spread exceeds {AM5_SPREAD_WARN}x; the machine " f"was not quiet and this number is directional only") print(" NOTE: reported, not gated — specs/GameKernel.md §5 declares AM-5") print(" `recorded not gated`. Promoting it is a spec change and") print(" needs an ADR; this tool does not promote itself.") return rc def self_test(): """Each assertion pins a failure this tool must detect.""" results = [] def check(name, cond, detail=""): results.append((name, cond, detail)) check("targets are numeric and positive", AM5_MAX_SECONDS > 0 and AM9_MAX_RSS_MB > 0, f"AM-5 <= {AM5_MAX_SECONDS:.0f}s, AM-9 <= {AM9_MAX_RSS_MB:.0f}MB") binary = test_binary() check("the release test binary is locatable", bool(binary), os.path.basename(binary) if binary else "NOT FOUND") if binary: # The control that matters: a filter matching no test must be # detected, not scored as a very low RSS. This is the AM-6/AM-7 # shape — a harness reporting a flattering number for work it # never did. _, ran = peak_rss_mb(binary, "no::such::test::name") check("a test that did not run is detected, not scored", not ran, "refusing to report RSS for work never done") rss, ran_real = peak_rss_mb(binary) check("the real workload runs and measures non-zero", ran_real and rss > 1.0, f"{rss:.1f} MB") # The control for the bug this tool actually had: cross-validate # against an independent implementation. getrusage(RUSAGE_CHILDREN) # reported 38.2 MB where the workload used 12.3 MB, and only a # second opinion revealed it. gnu = shutil.which("time") or "/usr/bin/time" if os.path.exists("/usr/bin/time"): r = subprocess.run(["/usr/bin/time", "-v", binary, AM9_TEST, "--exact"], cwd=ROOT, capture_output=True, text=True) m = re.search(r"Maximum resident set size \(kbytes\): (\d+)", r.stdout + r.stderr) if m: indep = int(m.group(1)) / 1024.0 near = abs(indep - rss) <= max(3.0, 0.25 * indep) check("RSS agrees with an independent measurement", near, f"ours {rss:.1f} MB vs /usr/bin/time {indep:.1f} MB") # The control for the defect this instrument actually had. check("AM-5 refuses to measure under load", AM5_MAX_LOAD_PER_CPU > 0 and "ABORT — load average" in open(__file__).read(), f"limit {AM5_MAX_LOAD_PER_CPU} per CPU; T03 reported 87.0 s under " f"contention against 54.1 s quiet") check("AM-5 takes the best of several samples", AM5_SAMPLES >= 3, f"{AM5_SAMPLES} samples") check("load is measured per CPU, not raw", 0.0 <= load_per_cpu() < 100.0, f"{load_per_cpu():.2f} per CPU now") check("clean build measures into a throwaway target dir, not target/", "CARGO_TARGET_DIR" in open(__file__).read() and "shutil.rmtree" in open(__file__).read(), "measuring must not destroy the working cache") print("runtime-metrics self-test (positive control)") ok = True for name, passed, det in results: print(f" [{'ok ' if passed else 'FAIL'}] {name}" + (f" — {det}" if det else "")) ok &= passed return 0 if ok else 1 def main(): enter_root() if "--self-test" in sys.argv: return self_test() return report(fast="--fast" in sys.argv) if __name__ == "__main__": sys.exit(main())