diff --git a/tests/test_compute_remeasure.py b/tests/test_compute_remeasure.py new file mode 100644 index 0000000..38e9982 --- /dev/null +++ b/tests/test_compute_remeasure.py @@ -0,0 +1,94 @@ +"""On-GPU throughput re-measurement gate (the airtight compute-forgery gate). + +op_compute_remeasure re-runs a would-be king's exact config on the validator's GPU and +rejects a declared throughput that can't be reproduced. The GPU probe itself needs +hardware (validated in shadow on the box), but the mode-parsing + the pure verdict +(GPU-scaled margin comparison) + fail-open behaviour are unit-tested here against the +real 2026-07 numbers. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import ralph_bootstrap # noqa: F401,E402 +from validator.validator import ( # noqa: E402 + _remeasure_mode, + _remeasure_verdict, + op_compute_remeasure, +) + +H100 = "NVIDIA H100 80GB HBM3" +H200 = "NVIDIA H200" + + +# ---------------------------------------------------------------- mode parsing +@pytest.mark.parametrize("val,expect", [ + (None, "off"), ("", "off"), ("0", "off"), ("off", "off"), + ("1", "enforce"), ("on", "enforce"), ("enforce", "enforce"), ("ENFORCE", "enforce"), + ("shadow", "shadow"), ("Shadow", "shadow"), +]) +def test_remeasure_mode(monkeypatch, val, expect): + if val is None: + monkeypatch.delenv("RALPH_COMPUTE_REMEASURE", raising=False) + else: + monkeypatch.setenv("RALPH_COMPUTE_REMEASURE", val) + assert _remeasure_mode() == expect + + +# --------------------------------------------------------------- verdict logic +def test_rejects_kaizen_forgery(): + # Kaizen b1930c27: declared 356k on H100; validator re-measures the same uncompiled + # Muon config at ~86k. Same GPU -> scale 1.0. ratio 4.1x >> 1.6 margin -> REJECT. + ok, detail = _remeasure_verdict(356_000, 86_000, H100, H100, margin=1.6) + assert not ok and "ratio 4.1x" in detail + + +def test_passes_danielortega_with_hardware_and_torchver_variance(): + # danielortega: declared 239k on H200; validator (H100) re-measures his COMPILED recipe + # at ~172k. peak scale H200/H100 == 1.0; margin 1.6 -> ceiling 275k. 239k < 275k -> PASS + # (absorbs the ~15% torch-version gap + H200 memory-bandwidth edge). + ok, _ = _remeasure_verdict(239_000, 172_000, H200, H100, margin=1.6) + assert ok + + +def test_gpu_scaling_up_for_slower_validator(): + # Miner on H200 (fast), validator on A100 (312 TF). A validator that re-measures 100k + # on its A100 scales UP by 989/312 = 3.17 -> ceiling 100k*3.17*1.6 = 507k. A declared + # 300k on H200 is achievable -> PASS (don't false-reject a genuinely-faster GPU). + ok, detail = _remeasure_verdict(300_000, 100_000, H200, "NVIDIA A100", margin=1.6) + assert ok and "scale 3.17" in detail + + +def test_rejects_gross_forgery_even_with_scaling(): + ok, _ = _remeasure_verdict(800_000, 100_000, H100, H100, margin=1.6) # 8x + assert not ok + + +def test_fail_open_when_no_measurement(): + ok, detail = _remeasure_verdict(300_000, 0, H100, H100, margin=1.6) + assert ok and "inconclusive" in detail + + +def test_margin_boundary(): + # exactly at the margin passes; just over rejects (same GPU, scale 1.0) + assert _remeasure_verdict(160_000, 100_000, H100, H100, margin=1.6)[0] + assert not _remeasure_verdict(160_001, 100_000, H100, H100, margin=1.6)[0] + + +# ------------------------------------------------------------- op fail-open/off +def test_op_off_is_disabled(monkeypatch, tmp_path): + monkeypatch.delenv("RALPH_COMPUTE_REMEASURE", raising=False) + ok, detail = op_compute_remeasure(tmp_path, tmp_path) + assert ok and "disabled" in detail + + +def test_op_skips_without_final_state(monkeypatch, tmp_path): + # enabled, but no final_state -> fail-open skip (never a false-reject on a glitch) + monkeypatch.setenv("RALPH_COMPUTE_REMEASURE", "1") + ok, detail = op_compute_remeasure(tmp_path, tmp_path) + assert ok and "no final_state" in detail diff --git a/validator/service.py b/validator/service.py index 513b00e..45442cf 100644 --- a/validator/service.py +++ b/validator/service.py @@ -45,7 +45,7 @@ from validator.hf_poller import DEFAULT_REPO as DEFAULT_HF_REPO from validator.hf_poller import poll_hub from validator.scoring import score_bundle -from validator.validator import judge_submission +from validator.validator import judge_submission, op_compute_remeasure # Bittensor's bt.logging hijacks Python's logging module and raises the root @@ -428,6 +428,19 @@ def score_and_decide( is_first = king is None and hwm is None decisively = score.decisively_beats_king or is_first + # Pre-crown on-GPU throughput re-measurement: run ONLY for a would-be king (cost is + # O(king-changes)). enforce withholds the crown when the declared throughput can't be + # reproduced on the validator's GPU; off/shadow never block (shadow logs the verdict). + # Fail-OPEN on every ambiguity; only PREVENTS a promotion, never dethrones the sitting king. + remeasure_withheld = False + if decisively: + ok_rm, detail_rm = op_compute_remeasure(RALPH_ROOT, bundle_dir) + result.operations["op_compute_remeasure"] = {"ok": ok_rm, "detail": detail_rm} + if not ok_rm: + remeasure_withheld = True + decisively = False + is_first = False + classification, weight_credit = _classify_outcome( decisively=decisively, val_bpb=result.hidden_eval.val_bpb, @@ -436,7 +449,9 @@ def score_and_decide( bundle_dir=bundle_dir, ) accepted = classification == "king_change" - if classification == "king_change": + if remeasure_withheld: + status = "remeasure_withheld" + elif classification == "king_change": status = "accepted" elif classification == "meaningful_failure": status = "meaningful_failure" @@ -466,6 +481,7 @@ def score_and_decide( "decisive": score.decisively_beats_king, "accepted": accepted, "is_first": is_first, + "remeasure_withheld": remeasure_withheld, "result": result, "score_report": score, } diff --git a/validator/validator.py b/validator/validator.py index 69476ce..6632e62 100644 --- a/validator/validator.py +++ b/validator/validator.py @@ -236,6 +236,188 @@ def _canonical_code_epoch() -> float | None: return None +# --- On-GPU throughput re-measurement (the airtight compute-forgery gate) -------- +# +# op1's MFU ceiling (check_compute_plausibility) is a static heuristic — it narrows the +# forgery window but cannot know a specific recipe's real achievable throughput. This gate +# MEASURES it: re-run the challenger's EXACT config for K steps on the validator's own GPU +# (synthetic data — throughput is data-independent) and reject if the DECLARED tok/s +# exceeds the measured achievable by more than a GPU-scaled margin. There is nothing to +# game: it is the physical rate the forger lies about. Expensive (GPU-minutes, incl. any +# torch.compile autotune) -> the CALLER runs it ONLY for a would-be king. off/shadow/ +# enforce via RALPH_COMPUTE_REMEASURE; every ambiguity fails OPEN (never a false-reject on +# a validator glitch — a failed verdict only WITHHOLDS a promotion, never dethrones). +def _remeasure_mode() -> str: + import os + v = os.environ.get("RALPH_COMPUTE_REMEASURE", "").strip().lower() + if v in ("1", "on", "true", "enforce"): + return "enforce" + if v == "shadow": + return "shadow" + return "off" + + +def _remeasure_verdict( + declared_tps: float, measured_tps: float, declared_gpu: str, + validator_gpu: str, margin: float, +) -> tuple[bool, str]: + """Pure verdict: is the declared throughput achievable given what the validator + re-measured on its OWN GPU? Scales the measurement to the miner's declared GPU by bf16 + peak (H100<->H200 == 1.0 by peak; the margin absorbs memory-bandwidth + torch-version + + thermal variance, e.g. danielortega's legit +15%). ok=True iff + declared <= measured * gpu_scale * margin. Unit-tested (no GPU).""" + from validator.integrity import _gpu_bf16_peak_flops + if measured_tps <= 0: + return True, "re-measure inconclusive (no measured throughput)" + gpu_scale = _gpu_bf16_peak_flops(declared_gpu) / _gpu_bf16_peak_flops(validator_gpu) + ceiling = measured_tps * gpu_scale * margin + ratio = declared_tps / measured_tps + detail = ( + f"declared {declared_tps:,.0f} tok/s vs re-measured {measured_tps:,.0f} on " + f"'{validator_gpu or 'validator-GPU'}' (declared '{declared_gpu or '?'}', " + f"scale {gpu_scale:.2f}, margin {margin:.1f}x -> ceiling {ceiling:,.0f}; ratio {ratio:.1f}x)" + ) + return declared_tps <= ceiling, detail + + +def _validator_gpu_name() -> str: + """The validator's own GPU name, for scaling the re-measurement. Best-effort.""" + import subprocess + try: + out = subprocess.run( + ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], + capture_output=True, text=True, timeout=10, + ) + if out.returncode == 0 and out.stdout.strip(): + return out.stdout.strip().splitlines()[0].strip() + except Exception: # noqa: BLE001 + pass + return "" + + +def _run_throughput_probe(recipe_dir: Path, proof_dir: Path, cfg: dict) -> float | None: + """Re-run the miner's config for K steps on synthetic data; return steady-state tok/s + (skipping step-0 warmup/compile), or None on any failure/timeout.""" + import os + import shutil + import subprocess + import tempfile + import time + + from proof.runner import _sanitized_env, apply_patch + K = int(os.environ.get("RALPH_REMEASURE_STEPS", "30")) + timeout_s = int(os.environ.get("RALPH_REMEASURE_TIMEOUT_S", "1500")) + with tempfile.TemporaryDirectory() as tmp: + tmp_p = Path(tmp) + workdir = tmp_p / "workdir" + for sub in ("model", "recipe", "data", "configs"): + src = recipe_dir / sub + if src.exists(): + shutil.copytree(src, workdir / sub, dirs_exist_ok=True) + patch = proof_dir / "patch.diff" + if patch.exists(): + try: + apply_patch(workdir, patch) + except Exception: # noqa: BLE001 + return None + env = _sanitized_env(extra={ + "PYTHONPATH": str(workdir), + "PYTORCH_CUDA_ALLOC_CONF": "expandable_segments:True", + }) + data_dir = workdir / "data" + try: # synthetic shards — throughput is data-independent + subprocess.run( + [sys.executable, "-m", "data.prepare", "--source", "synthetic", + "--out", str(data_dir / "shards"), "--shard-tokens", "5000000", + "--total-tokens", "40000000", + "--manifest", str(data_dir / "data_manifest.json")], + cwd=str(workdir), env=env, timeout=300, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + except (subprocess.SubprocessError, OSError): + return None + probe_cfg = dict(cfg) + probe_cfg["total_steps"] = K + probe_cfg["log_every"] = max(1, K // 5) + cfg_file = tmp_p / "probe_cfg.json" + cfg_file.write_text(json.dumps(probe_cfg)) + out_dir = tmp_p / "out" + out_dir.mkdir() + train_py = workdir / "recipe" / "train.py" + if not train_py.exists(): + return None + cmd = [ + sys.executable, str(train_py), "--config", str(cfg_file), + "--manifest", str(data_dir / "data_manifest.json"), + "--data-base-dir", str(data_dir), "--out-dir", str(out_dir), + ] + proc = subprocess.Popen(cmd, cwd=str(workdir), env=env, + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + rlog = out_dir / "training_log.jsonl" + start = time.time() + while time.time() - start < timeout_s: + if proc.poll() is not None: + break + time.sleep(3) + if proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=20) + except subprocess.TimeoutExpired: + proc.kill() + if not rlog.exists(): + return None + try: + rows = [json.loads(ln) for ln in rlog.read_text().splitlines() if ln.strip()] + except ValueError: + return None + warm = max(2, K // 5) # skip step-0 (compile) + warmup + steady = [r for r in rows + if int(r.get("step", 0)) >= warm and "elapsed_s" in r and "tokens_seen" in r] + if len(steady) < 2: + return None + dt = steady[-1]["elapsed_s"] - steady[0]["elapsed_s"] + dtok = steady[-1]["tokens_seen"] - steady[0]["tokens_seen"] + return dtok / dt if dt > 0 else None + + +def op_compute_remeasure(ralph_root: Path, proof_dir: Path) -> tuple[bool, str]: + """Pre-crown, would-beat-king-only compute re-measurement (see module comment). + off/shadow/enforce via RALPH_COMPUTE_REMEASURE; every ambiguity fails OPEN.""" + import os + mode = _remeasure_mode() + if mode == "off": + return True, "compute re-measure disabled" + from ralph_bootstrap import RECIPE_DIR + fs_path = proof_dir / "training" / "final_state.json" + if not fs_path.exists(): + return True, "re-measure skipped: no final_state" + try: + fs = json.loads(fs_path.read_text()) + except (OSError, ValueError): + return True, "re-measure skipped: unreadable final_state" + cfg = fs.get("config") or {} + try: + tokens = float(fs.get("tokens_seen", 0) or 0) + wall = float(fs.get("wall_clock_s", 0) or 0) + except (TypeError, ValueError): + return True, "re-measure skipped: non-numeric final_state" + if tokens <= 0 or wall <= 0 or not cfg: + return True, "re-measure skipped: incomplete final_state/config" + declared_tps = tokens / wall + declared_gpu = str(fs.get("gpu_name") or fs.get("device") or "") + measured_tps = _run_throughput_probe(RECIPE_DIR, proof_dir, cfg) + if measured_tps is None or measured_tps <= 0: + return True, "re-measure inconclusive (probe failed/timed out) -> skip" + margin = float(os.environ.get("RALPH_REMEASURE_MARGIN", "1.6") or 1.6) + ok, detail = _remeasure_verdict( + declared_tps, measured_tps, declared_gpu, _validator_gpu_name(), margin + ) + if mode == "shadow": + return True, f"[shadow] would_{'PASS' if ok else 'REJECT'}: {detail}" + return ok, (("compute re-measure OK: " if ok else "fabricated compute (re-measured): ") + detail) + + def op1_diff_and_integrity( ralph_root: Path, submission_payload: dict,