diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cdbc43f..b226a42 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,10 +21,17 @@ jobs: # The Ralph codebase imports from a sibling RalphLabsAI/recipe checkout # via ralph_bootstrap.py. Clone it into the parent directory so # RALPH_RECIPE_DIR autodetection finds it. + # + # Pin to the CANONICAL recipe tag (the one the validator runs), NOT recipe + # main: main tracks the latest king's auto-merged recipe, which can carry a + # model that breaks the protocol tests (e.g. a U-Net skip-gate forward that + # index-errors on the tests' small config). Bump this when the canonical + # recipe advances. - name: Checkout RalphLabsAI/recipe uses: actions/checkout@v4 with: repository: RalphLabsAI/recipe + ref: recipe-v0.2.3 path: recipe_repo - name: Set up Python ${{ matrix.python-version }} diff --git a/tests/test_compute_crown_gate.py b/tests/test_compute_crown_gate.py new file mode 100644 index 0000000..6c056a6 --- /dev/null +++ b/tests/test_compute_crown_gate.py @@ -0,0 +1,54 @@ +"""Compute-aware crown gate (calibrated net-score). A challenger must beat the +king on quality/benchmark AND be net-positive once its compute is charged against +the *calibrated* H100 reference. Rewards efficiency, rejects runaway compute, with +no hard hour cap. Fully tunable (RALPH_COMPUTE_COST_WEIGHT / _H100_MATMUL_MS_REF / +_CROWN_GATE).""" +from validator.scoring import ( + DEFAULT_H100_MATMUL_MS_REF, + _h100_matmul_ms_ref, + score_bundle, +) + + +def _score(wall_h, gain=0.05, matmul_ms=0.51, **kw): + # king at val_bpb 1.50 / benchmark 0.95; challenger improves val_bpb by `gain` + base = dict( + val_bpb=1.50 - gain, benchmark_accuracy=0.95, + king_val_bpb=1.50, king_benchmark=0.95, + noise_floor_margin=0.014, matmul_ms=matmul_ms, wall_clock_s=wall_h * 3600, + ) + base.update(kw) + return score_bundle(**base) + + +def test_calibrated_reference_not_placeholder(): + assert _h100_matmul_ms_ref() == DEFAULT_H100_MATMUL_MS_REF == 0.51 + + +def test_efficient_incremental_win_crowns(): + # 0.05 bpb gain at ~10 H100h -> net-positive -> crowns + r = _score(wall_h=10, gain=0.05) + assert r.decisively_beats_king and r.score > 0 + + +def test_runaway_compute_rejected_despite_quality_win(): + # same 0.05 gain at 100 H100h -> net-negative -> the gate blocks the crown + r = _score(wall_h=100, gain=0.05) + assert r.score < 0 and not r.decisively_beats_king + + +def test_gate_disabled_reverts_to_quality_only(monkeypatch): + monkeypatch.setenv("RALPH_COMPUTE_CROWN_GATE", "0") + r = _score(wall_h=100, gain=0.05) + assert r.decisively_beats_king # quality win crowns regardless of compute + + +def test_cost_weight_tunable_up(monkeypatch): + monkeypatch.setenv("RALPH_COMPUTE_COST_WEIGHT", "0.05") # aggressive pressure + r = _score(wall_h=10, gain=0.05) # 0.05 - 0.05*10 = -0.45 -> rejected + assert not r.decisively_beats_king + + +def test_reference_tunable(monkeypatch): + monkeypatch.setenv("RALPH_H100_MATMUL_MS_REF", "1.02") # 2x -> doubles cost + assert _h100_matmul_ms_ref() == 1.02 diff --git a/tests/test_op4_eval_cache.py b/tests/test_op4_eval_cache.py new file mode 100644 index 0000000..1c67b34 --- /dev/null +++ b/tests/test_op4_eval_cache.py @@ -0,0 +1,80 @@ +"""op4 hidden-eval result cache. A deferred challenger (king min-tenure guard) is +re-scored every epoch while it waits out the incumbent's tenure; the bundle and +the held-out eval shard are immutable across those epochs, so the ~90s GPU eval +is cached, keyed on the eval shard so it invalidates on eval rotation. Stored as +a per-bundle dotfile; op1 integrity is manifest-based so the extra file is ignored.""" +from pathlib import Path + +from eval.hidden_eval import HiddenEvalResult +from validator.validator import ( + _eval_cache_path, + _eval_shard_fingerprint, + _load_cached_hidden_eval, + _save_cached_hidden_eval, + op4_hidden_eval, +) + + +def _shard(eval_dir: Path, tokens: bytes = b"abc", bench: bytes = b"[]") -> None: + eval_dir.mkdir(parents=True, exist_ok=True) + (eval_dir / "active_tokens.bin").write_bytes(tokens) + (eval_dir / "active_benchmark.json").write_bytes(bench) + + +def _result(val_bpb: float = 1.5) -> HiddenEvalResult: + return HiddenEvalResult( + val_bpb=val_bpb, benchmark_accuracy=0.9, + tokens_evaluated=100, benchmark_examples=10, eval_set_hash="deadbeef", + ) + + +def test_cache_roundtrip_and_shard_invalidation(tmp_path): + eval_dir = tmp_path / "ralph" / "eval" / "private" + _shard(eval_dir) + proof = tmp_path / "queue" / "pending" / "abc123" + proof.mkdir(parents=True) + fp = _eval_shard_fingerprint(eval_dir) + _save_cached_hidden_eval(proof, fp, _result(1.5)) + got = _load_cached_hidden_eval(proof, fp) + assert got is not None and got.val_bpb == 1.5 and got.eval_set_hash == "deadbeef" + # rotate the eval shard -> fingerprint changes -> cache miss (no stale score) + _shard(eval_dir, tokens=b"ROTATED-SHARD") + fp2 = _eval_shard_fingerprint(eval_dir) + assert fp2 != fp + assert _load_cached_hidden_eval(proof, fp2) is None + + +def test_op4_short_circuits_on_cache_hit(tmp_path): + # Pre-populate the cache; op4 must return it even with NO checkpoint present, + # proving the cache is consulted before the (expensive) checkpoint load + eval. + ralph_root = tmp_path / "ralph" + _shard(ralph_root / "eval" / "private") + proof = tmp_path / "queue" / "pending" / "abc123" + (proof / "training").mkdir(parents=True) # deliberately NO checkpoint.pt + fp = _eval_shard_fingerprint(ralph_root / "eval" / "private") + _save_cached_hidden_eval(proof, fp, _result(1.42)) + ok, detail, res = op4_hidden_eval(ralph_root, proof) + assert ok is True and res is not None and res.val_bpb == 1.42 + assert "cached" in detail + + +def test_op4_missing_checkpoint_still_rejects_without_cache(tmp_path): + # No cache + no checkpoint -> normal rejection; the cache never masks errors. + ralph_root = tmp_path / "ralph" + _shard(ralph_root / "eval" / "private") + proof = tmp_path / "queue" / "pending" / "def456" + proof.mkdir(parents=True) + ok, detail, res = op4_hidden_eval(ralph_root, proof) + assert ok is False and res is None and "missing checkpoint" in detail + + +def test_cache_path_is_per_bundle_dotfile(tmp_path): + # Inside the bundle dir as a dotfile -> per-bundle (no cross-bundle collision) + # and ignored by op1 (manifest-based integrity verifies only declared files). + proof_a = tmp_path / "queue" / "pending" / "aaa" + proof_b = tmp_path / "queue" / "pending" / "bbb" + for p in (proof_a, proof_b): + p.mkdir(parents=True) + assert _eval_cache_path(proof_a) == proof_a / ".hidden_eval_cache.json" + assert _eval_cache_path(proof_a) != _eval_cache_path(proof_b) + assert _eval_cache_path(proof_a).name.startswith(".") diff --git a/tests/test_recovered_weights_no_double_king.py b/tests/test_recovered_weights_no_double_king.py new file mode 100644 index 0000000..cb96035 --- /dev/null +++ b/tests/test_recovered_weights_no_double_king.py @@ -0,0 +1,42 @@ +"""Regression: a rate-limited *recovered* weight must never resurrect a king the +throne has moved off of. Before the fix, the recovery merge re-added the previous +king at king-level weight on top of the new king's, so set_weights emitted to BOTH +kings every epoch until a set_weights finally landed and cleared pending_weights. +""" +from validator.service import KING_POOL_FRACTION, _merge_recovered_weights + + +def test_king_change_drops_stale_recovered_king(): + # NEW king crowned this epoch; the pending file still holds the OLD king at + # king-level weight from a previous rate-limited epoch. + round_scores = {"NEW_KING": 1.0} + recovered = {"OLD_KING": 1.0, "MF_MINER": 0.05} + out = _merge_recovered_weights(dict(round_scores), recovered, "NEW_KING") + assert "OLD_KING" not in out # dethroned king is NOT resurrected + assert out["NEW_KING"] == 1.0 # current king intact + assert out["MF_MINER"] == 0.05 # sub-king mf credit still recovered + + +def test_current_king_value_is_authoritative(): + # round_scores reflects a 90/10 split this epoch; a stale 1.0 recovered for + # the SAME king must not override the authoritative 0.9. + round_scores = {"KING": KING_POOL_FRACTION, "MF": 0.1} + recovered = {"KING": 1.0} + out = _merge_recovered_weights(dict(round_scores), recovered, "KING") + assert out["KING"] == KING_POOL_FRACTION + + +def test_meaningful_failure_recovery_preserved(): + # No king change; an mf credit that never landed last epoch is still recovered. + round_scores = {"KING": 1.0} + recovered = {"MF": 0.1} + out = _merge_recovered_weights(dict(round_scores), recovered, "KING") + assert out["MF"] == 0.1 + assert out["KING"] == 1.0 + + +def test_no_current_king_still_drops_king_level_recovered(): + # Throne cleared (genesis / post-reset): a king-level recovered weight from + # the prior reign must not sneak back in as a phantom king. + out = _merge_recovered_weights({}, {"OLD_KING": 1.0}, None) + assert out == {} diff --git a/validator/scoring.py b/validator/scoring.py index 5f1cc0c..c198bd0 100644 --- a/validator/scoring.py +++ b/validator/scoring.py @@ -35,6 +35,50 @@ # the published verdict at docs/direction_reframe/00_VERDICT.md §6. DOMINANT_QUALITY_MULTIPLIER = 3.0 +# Calibrated H100 matmul_ms reference for cost normalization. Measured on an +# NVIDIA H100 PCIe (float32) running the run_calibration workload (~0.51 ms). The +# previous 5.0 was a placeholder that inflated normalized H100-hours ~14x (it made +# a real ~7 wall-clock-hour run look like 102 "H100-hours"). Override per host +# with RALPH_H100_MATMUL_MS_REF (e.g. a different reference GPU). +DEFAULT_H100_MATMUL_MS_REF = 0.51 + + +def _h100_matmul_ms_ref() -> float: + try: + v = float(os.environ.get("RALPH_H100_MATMUL_MS_REF", "")) + except (TypeError, ValueError): + return DEFAULT_H100_MATMUL_MS_REF + return v if v > 0 else DEFAULT_H100_MATMUL_MS_REF + + +# bpb-points charged per normalized H100-hour in the net-score crown gate. +# Exchange rate: 1/weight = H100-hours of compute that one bpb-point of quality +# improvement "pays for". At 0.002 a typical 0.05-bpb incremental win stays +# net-positive up to ~25 H100h and is rejected beyond that — efficiency pressure +# without freezing incremental competition. Tune UP for more pressure. +DEFAULT_COMPUTE_COST_WEIGHT = 0.002 + + +def _compute_cost_weight() -> float: + """Weight on normalized H100-hours — the cost term in `score`, and thus in the + net-score crown gate. Tunable via RALPH_COMPUTE_COST_WEIGHT (default + DEFAULT_COMPUTE_COST_WEIGHT). Higher = more efficiency pressure (a wasteful run + needs proportionally larger quality/benchmark gains to stay net-positive).""" + try: + v = float(os.environ.get("RALPH_COMPUTE_COST_WEIGHT", "")) + except (TypeError, ValueError): + return DEFAULT_COMPUTE_COST_WEIGHT + return v if v >= 0 else DEFAULT_COMPUTE_COST_WEIGHT + + +def _compute_crown_gate_enabled() -> bool: + """Net-score crown gate. On by default now the cost reference is calibrated; + RALPH_COMPUTE_CROWN_GATE in {0,false,no,off} reverts to quality/benchmark-only + crowning (escape hatch if the cost model ever misbehaves on mainnet).""" + return os.environ.get("RALPH_COMPUTE_CROWN_GATE", "1").strip().lower() not in { + "0", "false", "no", "off", + } + def get_king_rule() -> str: """Return the currently-active king-selection rule. @@ -76,17 +120,19 @@ class ScoreReport: def _hours_to_normalized_h100( matmul_ms: float, wall_clock_s: float, - h100_matmul_ms_ref: float = 5.0, + h100_matmul_ms_ref: float | None = None, ) -> float: """Translate wall-clock into normalized H100-hours via the calibration benchmark's matmul timing. The miner's machine took (matmul_ms / h100_ref) times as long as an H100 would have, so each wall-clock hour counts as (h100_ref / matmul_ms) normalized H100-hours. - Reference: h100_matmul_ms_ref is the matmul_ms on an H100 for the - calibration workload, to be measured in Phase 0.5 and pinned in this file. - The 5.0 default is a placeholder. + Reference: h100_matmul_ms_ref is the matmul_ms on a reference H100 for the + calibration workload (DEFAULT_H100_MATMUL_MS_REF, calibrated on H100 PCIe), + overridable via RALPH_H100_MATMUL_MS_REF. """ + if h100_matmul_ms_ref is None: + h100_matmul_ms_ref = _h100_matmul_ms_ref() if matmul_ms <= 0: return wall_clock_s / 3600.0 speed_factor = h100_matmul_ms_ref / matmul_ms @@ -104,7 +150,7 @@ def score_bundle( tier: str = "verified", bpb_weight: float = 1.0, benchmark_weight: float = 1.0, - cost_weight: float = 0.1, + cost_weight: float | None = None, ) -> ScoreReport: """ Quality gain on val_bpb is computed as (king - challenger) since lower @@ -147,6 +193,8 @@ def score_bundle( (benchmark_gain > noise_floor_margin and quality_gain >= -noise_floor_margin) ) + if cost_weight is None: + cost_weight = _compute_cost_weight() cost_h100h = _hours_to_normalized_h100(matmul_ms, wall_clock_s) # v1.2: no α factor. cost_effective = cost_h100h @@ -157,6 +205,16 @@ def score_bundle( - cost_weight * cost_effective ) + # Compute-aware crown gate (calibrated net-score): a challenger that beats the + # king on quality/benchmark must ALSO be net-positive once compute is charged + # — quality_gain + benchmark_gain must outweigh cost_weight * normalized + # H100-hours (i.e. score > 0). Rewards efficiency, rejects runaway compute, + # with NO hard hour cap. `decisively` is only ever True against a real + # incumbent (genesis crowns via is_first in the caller), so the first king is + # never blocked. Escape hatch: RALPH_COMPUTE_CROWN_GATE=0. + if decisively and _compute_crown_gate_enabled() and score <= 0: + decisively = False + return ScoreReport( val_bpb=val_bpb, benchmark_accuracy=benchmark_accuracy, diff --git a/validator/service.py b/validator/service.py index 2ec6d7b..f3c5e4f 100644 --- a/validator/service.py +++ b/validator/service.py @@ -536,6 +536,32 @@ def _clear_pending_weights(chain) -> None: MEANINGFUL_FAILURE_POOL_FRACTION = 0.1 +def _merge_recovered_weights( + round_scores: dict[str, float], + recovered: dict[str, float], + current_king_hotkey: str | None, +) -> dict[str, float]: + """Merge weights recovered from a rate-limited previous epoch into this + epoch's round_scores WITHOUT resurrecting a stale king. + + `_apply_pool_split` already weights the CURRENT king authoritatively every + epoch, so a recovered king-level weight (>= KING_POOL_FRACTION) for any + hotkey OTHER than the current king is a previous reign whose set_weights got + rate-limited. Merging it would split emission across two kings — the + dethroned king keeps earning until a set_weights finally lands and clears the + pending file. We drop those, and skip the current king (its authoritative + weight is already in round_scores). Only sub-king (meaningful_failure) + credits that never landed are recovered, max-by-hotkey. + """ + for hk, w in recovered.items(): + if hk == current_king_hotkey: + continue # current king already weighted this epoch — don't override + if w >= KING_POOL_FRACTION: + continue # stale king from a prior reign — never resurrect + round_scores[hk] = max(round_scores.get(hk, 0.0), w) + return round_scores + + def _require_merged_king_pr() -> bool: """Opt-in policy: when setting weights, only emit to a king whose GitHub recipe PR is MERGED. @@ -1160,10 +1186,14 @@ def run_epoch( # If no meaningful_failures this epoch, king gets 100%. round_scores = _apply_pool_split(chain, king_change_hotkey, meaningful_failure_hotkeys) - # Merge any weights recovered from a previous rate-limited epoch so we - # don't lose credit for a miner whose set_weights got dropped last time. - for hk, w in recovered.items(): - round_scores[hk] = max(round_scores.get(hk, 0.0), w) + # Merge any weights recovered from a previous rate-limited epoch so we don't + # lose an unlanded meaningful_failure credit — but NEVER resurrect a king the + # throne has since moved off of (that splits emission across two kings). + current_king_hotkey = king_change_hotkey + if current_king_hotkey is None: + _sitting = chain.get_king() + current_king_hotkey = _sitting.miner_hotkey if _sitting is not None else None + round_scores = _merge_recovered_weights(round_scores, recovered, current_king_hotkey) # weights_set records whether the weight extrinsic actually landed this # epoch. It is INDEPENDENT of whether we build the audit report: the report diff --git a/validator/validator.py b/validator/validator.py index 8aa7d58..f659aaf 100644 --- a/validator/validator.py +++ b/validator/validator.py @@ -610,10 +610,85 @@ def _opt_str(key: str) -> str | None: ) +# --- Hidden-eval result cache ------------------------------------------------- +# A deferred challenger (king min-tenure guard) is re-scored EVERY epoch while it +# waits out the incumbent's tenure (~300 blocks). The bundle and the held-out +# eval shard are both immutable across those epochs, so the op4 GPU eval (~90 s) +# returns an identical result each time — pure waste that also blocks the GPU +# from processing new submissions. Cache it, keyed on a fingerprint of the eval +# shard so the cache auto-invalidates the moment the shard is rotated. Stored as +# a dotfile inside the bundle dir; op1 integrity is manifest-based (verifies only +# the declared files), so the extra file is ignored. +_EVAL_CACHE_FIELDS = ( + "val_bpb", "benchmark_accuracy", "tokens_evaluated", "benchmark_examples", + "eval_set_hash", "val_seq_len", "sealed_stream_manifest_hash", "tail_val_bpb", +) + + +def _eval_shard_fingerprint(eval_dir: Path) -> str: + """sha256 over the held-out eval shard (tokens + benchmark). Changes iff the + eval set is rotated — exactly when a cached score MUST be discarded.""" + h = hashlib.sha256() + for name in ("active_tokens.bin", "active_benchmark.json"): + p = eval_dir / name + h.update(name.encode("utf-8")) + h.update(b"\x00") + h.update(p.read_bytes() if p.exists() else b"") + return h.hexdigest() + + +def _eval_cache_path(proof_dir: Path) -> Path: + # A dotfile INSIDE the bundle dir. op1 integrity is manifest-based (it verifies + # only the declared files — checkpoint/training_log/calibration/attestation/ + # patch — and recomputes bundle_hash from those four), so this extra file is + # ignored. It is archived with the bundle (harmless) and absent on a fresh + # re-download -> correct re-eval. Per-bundle, so no cross-bundle collision. + return proof_dir / ".hidden_eval_cache.json" + + +def _load_cached_hidden_eval(proof_dir: Path, shard_fp: str) -> HiddenEvalResult | None: + try: + d = json.loads(_eval_cache_path(proof_dir).read_text()) + except (FileNotFoundError, json.JSONDecodeError, OSError, ValueError): + return None + if d.get("eval_shard_fingerprint") != shard_fp: + return None # eval shard rotated since this was cached + r = d.get("result") + if not isinstance(r, dict): + return None + try: + return HiddenEvalResult(**{k: r[k] for k in _EVAL_CACHE_FIELDS if k in r}) + except (TypeError, KeyError): + return None + + +def _save_cached_hidden_eval(proof_dir: Path, shard_fp: str, result: HiddenEvalResult) -> None: + # A downstream (CSDP) report is a nested object we don't round-trip here — + # skip the cache rather than drop it; the next epoch re-evals. + if getattr(result, "downstream", None) is not None: + return + payload = { + "eval_shard_fingerprint": shard_fp, + "result": {k: getattr(result, k) for k in _EVAL_CACHE_FIELDS}, + } + try: + p = _eval_cache_path(proof_dir) + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(payload)) + except OSError: + pass # cache is a pure optimization — never fail scoring on a write error + + def op4_hidden_eval( ralph_root: Path, proof_dir: Path, ) -> tuple[bool, str, HiddenEvalResult | None]: + eval_dir = ralph_root / "eval" / "private" + shard_fp = _eval_shard_fingerprint(eval_dir) + cached = _load_cached_hidden_eval(proof_dir, shard_fp) + if cached is not None: + return True, f"val_bpb={cached.val_bpb:.4f} bench={cached.benchmark_accuracy:.3f} (cached)", cached + ckpt_path = proof_dir / "training" / "checkpoint.pt" if not ckpt_path.exists(): return False, f"missing checkpoint at {ckpt_path}", None @@ -638,11 +713,15 @@ def op4_hidden_eval( # Retry under the patched workdir so the actually-trained model code # is what scores the checkpoint. if _is_state_dict_shape_mismatch(e): - return _patched_hidden_eval(ralph_root, proof_dir, ckpt_path) + ok, detail, result = _patched_hidden_eval(ralph_root, proof_dir, ckpt_path) + if ok and result is not None: + _save_cached_hidden_eval(proof_dir, shard_fp, result) + return ok, detail, result raise if torch.cuda.is_available(): model = model.cuda() - result = run_hidden_eval(model, ralph_root / "eval" / "private", seq_len=cfg.max_seq_len // 2) + result = run_hidden_eval(model, eval_dir, seq_len=cfg.max_seq_len // 2) + _save_cached_hidden_eval(proof_dir, shard_fp, result) return True, f"val_bpb={result.val_bpb:.4f} bench={result.benchmark_accuracy:.3f}", result