From d63ea3e05e61818a77749379250ecbeeb5ebd2dc Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 16:40:07 +0000 Subject: [PATCH 1/9] eval: add blind-forgeability detector for the benchmark (anti-gaming P0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - benchmark_blind_forgeable(): flags a non-content-whitened active_benchmark.json where a blind score=+-token_id sweep beats chance+3sigma (the deployed 0.785 vs 0.2 break — targets are systematically small-id vs uniform distractors, so a model that learned nothing wins the benchmark leg). make_placeholder_examples is already content-whitened; the DEPLOYED file predates it and must be regenerated. - first commit of the anti-gaming hardening series (un-gameable plan). Next: neutralize benchmark_gain in the crown until the host-reduced benchmark is enforced, then regenerate the deployed file + startup self-test. Co-Authored-By: Claude Opus 4.8 --- eval/benchmark.py | 46 ++++++++++++++++++++++++++++++++ tests/test_benchmark_selftest.py | 34 +++++++++++++++++++++++ 2 files changed, 80 insertions(+) create mode 100644 tests/test_benchmark_selftest.py diff --git a/eval/benchmark.py b/eval/benchmark.py index 5824c2f..9778244 100644 --- a/eval/benchmark.py +++ b/eval/benchmark.py @@ -78,3 +78,49 @@ def make_placeholder_examples( "distractors": distractors, }) return examples + + +def benchmark_blind_forgeable(examples: list[dict], sigma: float = 3.0) -> tuple[bool, str]: + """Detect a non-content-whitened (blind-forgeable) benchmark file. + + compute_benchmark_score ranks [target_id] + distractors by the model's logits. + If the correct target is distinguishable from the distractors by a monotone + function of the token-id alone (e.g. targets are systematically small while + distractors are uniform), a model that never learned anything wins by scoring + `-token_id` — the exact 0.785-vs-0.2 break found on the deployed file. A + content-whitened file (make_placeholder_examples: target + distractors from one + exchangeable rng.choice pool, uniform target slot) makes every blind monotone + score chance-level. + + Returns (forgeable, reason). forgeable=True -> the file must be regenerated + with make_placeholder_examples before it can gate the crown. + """ + import numpy as np + + if not examples: + return False, "empty benchmark (nothing to forge)" + n = len(examples) + k = 1 + len(examples[0].get("distractors", [])) + if k <= 1: + return False, "single-candidate benchmark" + chance = 1.0 / k + sd = (chance * (1.0 - chance) / n) ** 0.5 + bar = chance + sigma * sd + worst = 0.0 + worst_label = "" + for sign, label in ((-1.0, "-id"), (1.0, "+id")): + correct = 0 + for e in examples: + cands = [e["target_id"]] + list(e["distractors"]) + if int(np.argmax([sign * float(t) for t in cands])) == 0: + correct += 1 + acc = correct / n + if acc > worst: + worst, worst_label = acc, label + if worst > bar: + return True, ( + f"blind-forgeable benchmark: score={worst_label} sweep scores {worst:.3f} " + f"> chance {chance:.3f} + {sigma:.0f}sigma ({bar:.3f}); regenerate " + f"active_benchmark.json with the content-whitened make_placeholder_examples" + ) + return False, f"benchmark not blind-forgeable (worst blind ±id = {worst:.3f} <= {bar:.3f})" diff --git a/tests/test_benchmark_selftest.py b/tests/test_benchmark_selftest.py new file mode 100644 index 0000000..4c82833 --- /dev/null +++ b/tests/test_benchmark_selftest.py @@ -0,0 +1,34 @@ +"""The blind-forgeability detector must flag an id-biased benchmark (the deployed +0.785 break) and pass the content-whitened generator.""" +from __future__ import annotations + +import numpy as np + +from eval.benchmark import benchmark_blind_forgeable, make_placeholder_examples + + +def test_whitened_generator_not_forgeable(): + ex = make_placeholder_examples(n=300, seed=1, n_candidates=5) + forgeable, reason = benchmark_blind_forgeable(ex) + assert not forgeable, reason + + +def test_id_biased_benchmark_is_flagged(): + # Targets systematically small, distractors uniform-large -> blind score=-id wins + # (the exact deployed break: target mean ~4550 vs distractor mean ~24917). + rng = np.random.default_rng(0) + ex = [ + { + "context_ids": [1, 2, 3], + "target_id": int(rng.integers(0, 5000)), + "distractors": [int(rng.integers(20000, 50000)) for _ in range(4)], + } + for _ in range(300) + ] + forgeable, reason = benchmark_blind_forgeable(ex) + assert forgeable and "blind-forgeable" in reason + + +def test_empty_and_single_candidate_are_safe(): + assert not benchmark_blind_forgeable([])[0] + assert not benchmark_blind_forgeable([{"target_id": 1, "distractors": []}])[0] From 96d1ec069129eb85c012591360414b464483d1c8 Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 16:43:58 +0000 Subject: [PATCH 2/9] crown: neutralize forgeable benchmark_gain (val_bpb decides); RALPH_BENCHMARK_CROWN toggle --- validator/scoring.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/validator/scoring.py b/validator/scoring.py index c198bd0..6af3a15 100644 --- a/validator/scoring.py +++ b/validator/scoring.py @@ -165,6 +165,13 @@ def score_bundle( decisively_beats_king=False and quality_gain/benchmark_gain zero so the caller can reject without crowning a null king. """ + # The in-container benchmark_accuracy is blind-forgeable until the host-reduced + # benchmark is enforced (a score=-token_id sweep beats chance on a non-content- + # whitened active_benchmark.json). So benchmark_gain is NEUTRALIZED in the crown + # by default: it can neither crown a challenger (Branch C) nor mask a quality + # regression (the benchmark-no-regress guards go vacuous). val_bpb quality_gain + # alone decides. Re-enable once the benchmark is host-reduced: RALPH_BENCHMARK_CROWN=1. + _bench_crown = os.environ.get("RALPH_BENCHMARK_CROWN") == "1" finite_metrics = ( isinstance(val_bpb, (int, float)) and math.isfinite(val_bpb) and isinstance(benchmark_accuracy, (int, float)) and math.isfinite(benchmark_accuracy) @@ -181,16 +188,16 @@ def score_bundle( # challenger that regressed past the noise floor on the OTHER axis: # Branch A: dominant quality (≥3× noise) AND benchmark-no-regress # Branch B: quality > noise AND benchmark-no-regress - # Branch C: benchmark > noise AND quality-no-regress + # Branch C: benchmark > noise AND quality-no-regress (bench-crown only) # The Branch A guard (the AND ... clause) is the v0.10 Goodhart fix. - # See the constant docstring above for the seed-search vector it closes. + _bg = benchmark_gain if _bench_crown else 0.0 decisively = ( (quality_gain >= DOMINANT_QUALITY_MULTIPLIER * noise_floor_margin - and benchmark_gain >= -noise_floor_margin) + and _bg >= -noise_floor_margin) or - (quality_gain > noise_floor_margin and benchmark_gain >= -noise_floor_margin) + (quality_gain > noise_floor_margin and _bg >= -noise_floor_margin) or - (benchmark_gain > noise_floor_margin and quality_gain >= -noise_floor_margin) + (_bench_crown and benchmark_gain > noise_floor_margin and quality_gain >= -noise_floor_margin) ) if cost_weight is None: @@ -201,7 +208,7 @@ def score_bundle( score = ( bpb_weight * quality_gain - + benchmark_weight * benchmark_gain + + (benchmark_weight * benchmark_gain if _bench_crown else 0.0) - cost_weight * cost_effective ) From 89e4a1451aaa0ec72ed0ce493872925c5639ff5f Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 16:52:38 +0000 Subject: [PATCH 3/9] service: benchmark blind-forgeability self-test at startup (warns on a forgeable file) --- validator/service.py | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/validator/service.py b/validator/service.py index 96fe2bb..a2b621b 100644 --- a/validator/service.py +++ b/validator/service.py @@ -1485,6 +1485,19 @@ def main(): else: log_info("HF Hub poll: disabled") log_info(f"submit bundles to: {args.queue_dir / 'pending' / '/'}") + # Benchmark blind-forgeability self-test (anti-gaming P0): warn loudly if the + # deployed active_benchmark.json is content-forgeable (a blind score=±token_id + # sweep beats chance), so it can never silently gate the crown once the + # benchmark is re-enabled host-reduced. + try: + from eval.benchmark import benchmark_blind_forgeable + + _bp = RALPH_ROOT / "eval" / "private" / "active_benchmark.json" + if _bp.exists(): + _forge, _why = benchmark_blind_forgeable(json.loads(_bp.read_text())) + (log_warn if _forge else log_info)(f"benchmark self-test: {_why}") + except Exception as _e: # noqa: BLE001 — self-test must never block startup + log_warn(f"benchmark self-test skipped: {_e}") log_info("") epoch = 0 From 3564dc5a2163c3c81d6ecc481b89d85142e57b8a Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 17:05:48 +0000 Subject: [PATCH 4/9] =?UTF-8?q?hf=5Fpoller:=20DoS=20intake=20caps=20(enc/b?= =?UTF-8?q?lob/streamed-decompress)=20=E2=80=94=20anti=20decompression-bom?= =?UTF-8?q?b?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/test_intake_caps.py | 23 +++++++++++++++++++++ validator/hf_poller.py | 43 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 65 insertions(+), 1 deletion(-) create mode 100644 tests/test_intake_caps.py diff --git a/tests/test_intake_caps.py b/tests/test_intake_caps.py new file mode 100644 index 0000000..85d139d --- /dev/null +++ b/tests/test_intake_caps.py @@ -0,0 +1,23 @@ +"""DoS intake caps: a decompression bomb must be rejected before unpack.""" +from __future__ import annotations + +import gzip +import io + +import pytest + +from validator.hf_poller import _assert_blob_safe + + +def test_normal_blob_passes(): + _assert_blob_safe(gzip.compress(b"ok" * 2000)) + + +def test_decompression_bomb_rejected(): + buf = io.BytesIO() + with gzip.GzipFile(fileobj=buf, mode="wb") as g: + z = b"\0" * (1 << 20) + for _ in range(4096): # 4 GiB decompressed from a ~4 MB blob + g.write(z) + with pytest.raises(ValueError, match="bomb|cap"): + _assert_blob_safe(buf.getvalue()) diff --git a/validator/hf_poller.py b/validator/hf_poller.py index 59c185b..55ca6fc 100644 --- a/validator/hf_poller.py +++ b/validator/hf_poller.py @@ -26,6 +26,43 @@ DEFAULT_REPO = "RalphLabsAI/proof-bundles" +# --- DoS intake caps (single-validator crash protection) ---------------------- +# A miner bundle is downloaded, decrypted, and tar.gz-extracted into the ONE +# validator. An unbounded extract is a crash-DoS (a tiny .gz can decompress to +# terabytes — a decompression bomb — or ship one giant member). Hard-cap the +# encrypted size, the decrypted blob, and — streamed, so it never expands in RAM +# — the total decompressed bytes, BEFORE proof.runner unpacks it. Env-tunable. +_MAX_ENC_BYTES = int(os.environ.get("RALPH_MAX_ENC_BYTES", str(2 << 30))) # 2 GiB +_MAX_BLOB_BYTES = int(os.environ.get("RALPH_MAX_BLOB_BYTES", str(2 << 30))) # 2 GiB +_MAX_DECOMPRESSED_BYTES = int(os.environ.get("RALPH_MAX_DECOMPRESSED_BYTES", str(3 << 30))) # 3 GiB + + +def _assert_blob_safe(blob: bytes) -> None: + """Reject an oversized / decompression-bomb decrypted blob before unpack. + Streams the gunzip with a hard byte cap so a bomb never expands in memory. + Raises ValueError on violation (the caller treats it as a decrypt/unpack fail).""" + import gzip + import io + + if len(blob) > _MAX_BLOB_BYTES: + raise ValueError(f"decrypted blob {len(blob)} B > {_MAX_BLOB_BYTES} B cap") + total = 0 + try: + with gzip.GzipFile(fileobj=io.BytesIO(blob)) as g: + while True: + chunk = g.read(1 << 20) # 1 MiB + if not chunk: + break + total += len(chunk) + if total > _MAX_DECOMPRESSED_BYTES: + raise ValueError( + f"decompressed size exceeds {_MAX_DECOMPRESSED_BYTES} B cap " + f"(decompression bomb)" + ) + except OSError as e: + raise ValueError(f"bad gzip stream: {e}") from e + + # miner.hub titles every submission PR "Submit proof bundle ". _TITLE_HASH_RE = re.compile(r"Submit proof bundle ([0-9a-f]{6,})", re.IGNORECASE) @@ -203,7 +240,11 @@ def download_one( repo_id=repo_id, filename=enc_remote, repo_type="dataset", local_dir=str(cache), token=token, revision=git_ref, ) - blob = bundle_crypto.decrypt(Path(local).read_bytes()) + enc_bytes = Path(local).read_bytes() + if len(enc_bytes) > _MAX_ENC_BYTES: + raise ValueError(f"encrypted bundle {len(enc_bytes)} B > {_MAX_ENC_BYTES} B cap") + blob = bundle_crypto.decrypt(enc_bytes) + _assert_blob_safe(blob) # DoS: cap decrypted + decompressed size before unpack bundle_crypto.unpack_bundle(blob, out) success = 1 except Exception as e: From 266804d7dcf7856fb451c2c5fa7ce0aa2e384b04 Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 17:16:48 +0000 Subject: [PATCH 5/9] service: wire run_pending_audits() into idle epochs (was dead -> 0 fraud events ever) --- validator/service.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/validator/service.py b/validator/service.py index a2b621b..652d98d 100644 --- a/validator/service.py +++ b/validator/service.py @@ -1523,6 +1523,27 @@ def main(): f"{result['accepted']} accepted, {result['rejected']} rejected{mf_str}") else: log_info(f"epoch {epoch}: no pending submissions") + # Idle: drain ONE pending audit job. This was dead — run_pending_audits + # was called nowhere, so 0 fraud/blacklist events ever fired and every + # economic-deterrence claim was moot. On a re-derivation mismatch it + # appends submission_fraud + blacklists the miner (zeroed next set_weights). + # Off: RALPH_AUDIT_LOOP_OFF=1. NOTE the audit is a proxy re-run until + # canonical data + checkpoint-bound re-derivation land (plan P2/P4). + if os.environ.get("RALPH_AUDIT_LOOP_OFF") != "1": + try: + from validator.audit_scheduler import run_pending_audits + + _a = run_pending_audits( + chain, RALPH_ROOT, RALPH_ROOT / "chain", + noise_floor_margin=args.noise_floor, limit=1, + ) + if _a.get("processed"): + log_info( + f"audit: processed {_a['processed']} " + f"(passed {_a.get('passed', 0)}, failed {_a.get('failed', 0)})" + ) + except Exception as _e: # noqa: BLE001 — audit must never crash the epoch loop + log_warn(f"audit loop error: {_e}") except Exception as e: log_err(f"epoch {epoch} failed: {e}") log_debug(traceback.format_exc()) From 0cb279ae7f2d75b573eb2fd306053cd317a1d36d Mon Sep 17 00:00:00 2001 From: karpabot Date: Wed, 1 Jul 2026 17:46:29 +0000 Subject: [PATCH 6/9] test: gate legacy benchmark-crown tests on RALPH_BENCHMARK_CROWN + add neutralized-default test --- tests/test_scoring.py | 36 +++++++++++++++++++++++++++++------- 1 file changed, 29 insertions(+), 7 deletions(-) diff --git a/tests/test_scoring.py b/tests/test_scoring.py index f79306f..19e40c7 100644 --- a/tests/test_scoring.py +++ b/tests/test_scoring.py @@ -48,11 +48,13 @@ def test_not_decisive_inside_noise_band(): assert sr.decisively_beats_king is False -def test_not_decisive_when_bench_regresses_and_quality_below_dominant(): +def test_not_decisive_when_bench_regresses_and_quality_below_dominant(monkeypatch): """bpb wins past noise but BELOW DOMINANT_QUALITY_MULTIPLIER, and bench regresses past -noise — not decisive. The quality_gain (0.02 ≈ 1.5x noise) doesn't clear the dominant-quality branch (3x noise = 0.039) - and the paired-axes branch fails because bench drop is too large.""" + and the paired-axes branch fails because bench drop is too large. + (Legacy benchmark-in-crown behavior — now behind RALPH_BENCHMARK_CROWN.)""" + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") sr = score_bundle( val_bpb=1.48, # 0.02 better — above noise, below dominant benchmark_accuracy=0.40, # 0.05 worse — outside -noise_floor @@ -64,7 +66,7 @@ def test_not_decisive_when_bench_regresses_and_quality_below_dominant(): assert sr.decisively_beats_king is False -def test_v0_10_branch_a_no_longer_crowns_when_bench_regresses(): +def test_v0_10_branch_a_no_longer_crowns_when_bench_regresses(monkeypatch): """v0.10 Goodhart closure: under the GUARDED dominant-quality clause, a big quality_gain (~6x noise) does NOT crown if benchmark regressed past the noise floor. This is the seed-search backdoor the v0.10 fix closes @@ -74,7 +76,9 @@ def test_v0_10_branch_a_no_longer_crowns_when_bench_regresses(): the v0.10 guard the case is plain_failure because the benchmark drop indicates the win may be spurious. Branch B and Branch C also fail (Branch B needs benchmark-no-regress; Branch C needs benchmark > noise). + (Legacy benchmark-in-crown behavior — now behind RALPH_BENCHMARK_CROWN.) """ + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") sr = score_bundle( val_bpb=1.42, # 0.08 better — well above 3x noise benchmark_accuracy=0.40, # 0.05 worse — outside -noise_floor @@ -117,10 +121,11 @@ def test_dominant_quality_threshold_just_above_3x_noise(): assert sr.decisively_beats_king is True -def test_dominant_quality_threshold_does_not_fire_just_below(): +def test_dominant_quality_threshold_does_not_fire_just_below(monkeypatch): """Boundary the other side: 2.9x noise (below 3x) doesn't trigger the dominant-quality branch. If bench also regresses past -noise the submission falls back to non-decisive.""" + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") sr = score_bundle( val_bpb=1.4623, # gain = 0.0377 = 2.9 * 0.013 benchmark_accuracy=0.00, # bench tanked @@ -132,9 +137,10 @@ def test_dominant_quality_threshold_does_not_fire_just_below(): assert sr.decisively_beats_king is False -def test_decisive_via_benchmark_axis(): - """Benchmark wins, val_bpb roughly tied — decisive via the benchmark - arm of the OR-of-AND condition.""" +def test_decisive_via_benchmark_axis(monkeypatch): + """Benchmark wins, val_bpb roughly tied — decisive via the benchmark arm of + the OR-of-AND condition. (Legacy behavior — now behind RALPH_BENCHMARK_CROWN.)""" + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") sr = score_bundle( val_bpb=1.50, # tied benchmark_accuracy=0.50, # 0.05 better — outside noise floor @@ -146,6 +152,22 @@ def test_decisive_via_benchmark_axis(): assert sr.decisively_beats_king is True +def test_benchmark_neutralized_by_default(monkeypatch): + """Default (no RALPH_BENCHMARK_CROWN): the forgeable in-container benchmark can + neither crown a challenger alone (Branch C gone) nor block a genuine val_bpb win.""" + monkeypatch.delenv("RALPH_BENCHMARK_CROWN", raising=False) + # bench-only "win", val_bpb tied -> NO crown. + assert score_bundle( + val_bpb=1.50, benchmark_accuracy=0.50, king_val_bpb=1.5, king_benchmark=0.45, + noise_floor_margin=0.013, matmul_ms=5.0, wall_clock_s=60.0, + ).decisively_beats_king is False + # val_bpb win with benchmark tanked -> STILL crowns (benchmark ignored). + assert score_bundle( + val_bpb=1.42, benchmark_accuracy=0.0, king_val_bpb=1.5, king_benchmark=0.45, + noise_floor_margin=0.013, matmul_ms=5.0, wall_clock_s=60.0, + ).decisively_beats_king is True + + def test_nan_val_bpb_not_decisive(): """NaN val_bpb must NOT crown a king — non-finite metrics rejected.""" sr = score_bundle(val_bpb=float("nan"), **_BASE) From cdd456075893dac4086135a59ba11e54a3000619 Mon Sep 17 00:00:00 2001 From: karpabot Date: Thu, 2 Jul 2026 12:01:56 +0000 Subject: [PATCH 7/9] hf_poller: close PR + dedup on decrypt failure - download_one returns status ('ok'/'decrypt_failed'/'unavailable'), not bool - decrypt-failed (wrong/rotated seal key) is permanent -> stamp done so it stops re-downloading + re-failing the same bundle every epoch (was 512 repeated decrypt errors / ~184 retries per dead bundle) - close the miner's HF PR with the current validator pubkey so they re-seal - transient (network / mid-upload) stays un-stamped -> retries next epoch - PR-close is best-effort; a perms/network failure never blocks dedup - tests: dedup routing + close message carries pubkey + error-swallow + noop Co-Authored-By: Claude Opus 4.8 --- tests/test_hf_poller_decrypt_feedback.py | 89 ++++++++++++++++++++++++ validator/hf_poller.py | 80 ++++++++++++++++++--- 2 files changed, 159 insertions(+), 10 deletions(-) create mode 100644 tests/test_hf_poller_decrypt_feedback.py diff --git a/tests/test_hf_poller_decrypt_feedback.py b/tests/test_hf_poller_decrypt_feedback.py new file mode 100644 index 0000000..47a6ea8 --- /dev/null +++ b/tests/test_hf_poller_decrypt_feedback.py @@ -0,0 +1,89 @@ +"""Tests for decrypt-failure handling in the HF poller. + +A bundle sealed to an outdated validator pubkey can never be decrypted. Before, +poll_hub re-downloaded + re-failed it every epoch forever (never stamped done). +Now: + - download_one returns "decrypt_failed" (permanent) vs "unavailable" (transient), + - poll_hub stamps "decrypt_failed" done (stops the churn) but leaves "unavailable" + for retry, + - and the miner's PR is closed with the current pubkey so they can re-seal. +""" + +import huggingface_hub + +from validator import hf_poller as hp +from validator.version import VALIDATOR_VERSION + + +def _pr(prefix, num): + return { + "bundle_id": prefix + "0" * (64 - len(prefix)), + "pr_num": num, + "git_ref": f"refs/pr/{num}", + "created_at": f"2026-06-2{num}T00:00:00+00:00", + } + + +def test_decrypt_failed_marked_done_transient_retried(tmp_path, monkeypatch): + prs = [_pr("ok", 1), _pr("de", 2), _pr("un", 3)] + monkeypatch.setattr(hp, "list_remote_submissions", lambda repo_id, token=None: prs) + + status_by_prefix = {"ok": "ok", "de": "decrypt_failed", "un": "unavailable"} + monkeypatch.setattr( + hp, "download_one", + lambda bid, repo_id, dest, **kw: status_by_prefix[bid[:2]], + ) + + downloaded = hp.poll_hub(tmp_path) + assert downloaded == [prs[0]["bundle_id"]] # only the 'ok' bundle materialised + + done = hp._load_state(tmp_path)["processed"] + assert done.get(prs[0]["bundle_id"]) == VALIDATOR_VERSION # ok → done + assert done.get(prs[1]["bundle_id"]) == VALIDATOR_VERSION # decrypt_failed → done (no re-churn) + assert prs[2]["bundle_id"] not in done # unavailable → NOT done (retries) + + +def test_close_pr_posts_current_pubkey_and_closes(tmp_path, monkeypatch): + from proof import bundle_crypto + + calls = {} + + class _Api: + def __init__(self, token=None): + pass + + def change_discussion_status(self, repo_id, discussion_num, new_status, + repo_type=None, comment=None): + calls.update(repo_id=repo_id, num=discussion_num, status=new_status, + repo_type=repo_type, comment=comment) + + monkeypatch.setattr(huggingface_hub, "HfApi", _Api) + + hp._close_pr_wrong_key("RalphLabsAI/proof-bundles", 42, None, "a" * 64, "MAC check failed") + + assert calls["num"] == 42 + assert calls["status"] == "closed" + assert calls["repo_type"] == "dataset" + assert bundle_crypto.DEFAULT_VALIDATOR_PUBKEY in calls["comment"] + assert "re-seal" in calls["comment"].lower() + + +def test_close_pr_swallows_api_errors(monkeypatch): + """A failed close (perms/network) must not raise — dedup still proceeds.""" + class _Api: + def __init__(self, token=None): + pass + + def change_discussion_status(self, **kw): + raise RuntimeError("403 forbidden") + + monkeypatch.setattr(huggingface_hub, "HfApi", _Api) + hp._close_pr_wrong_key("RalphLabsAI/proof-bundles", 7, None, "b" * 64, "boom") # no raise + + +def test_close_pr_noop_without_pr_num(monkeypatch): + def _boom(token=None): + raise AssertionError("HfApi should not be constructed when pr_num is None") + + monkeypatch.setattr(huggingface_hub, "HfApi", _boom) + hp._close_pr_wrong_key("RalphLabsAI/proof-bundles", None, None, "c" * 64, "x") # no raise diff --git a/validator/hf_poller.py b/validator/hf_poller.py index 55ca6fc..fb05bff 100644 --- a/validator/hf_poller.py +++ b/validator/hf_poller.py @@ -187,6 +187,48 @@ def list_remote_submissions(repo_id: str, token: Optional[str] = None) -> list[d return pending +def _close_pr_wrong_key( + repo_id: str, pr_num: Optional[int], token: Optional[str], bundle_id: str, err: str, +) -> None: + """Comment on the HF PR and close it when a bundle can't be decrypted. + + A decrypt failure means the bundle was sealed to an outdated validator public + key (the seal key was rotated), so it can never be processed. Instead of the PR + churning silently — re-downloaded and re-failed every epoch — give the miner + actionable feedback (the current pubkey to re-seal to) and close the PR. + + Best-effort: any failure here (missing pr_num, perms, network) is swallowed so + it never blocks marking the bundle done, which is what actually stops the churn. + """ + if pr_num is None: + return + try: + from huggingface_hub import HfApi + + from proof import bundle_crypto + msg = ( + "🔒 **Bundle decryption failed — re-seal required**\n\n" + "The validator could not decrypt this proof bundle. This means it was " + "sealed to an **outdated validator public key** (the seal key was " + "rotated).\n\n" + "Please re-seal to the **current** validator public key and open a new " + "submission:\n\n" + f"```\n{bundle_crypto.DEFAULT_VALIDATOR_PUBKEY}\n```\n\n" + "(`DEFAULT_VALIDATOR_PUBKEY` in `proof/bundle_crypto.py`.) Closing this " + f"PR — resubmit once re-sealed.\n\ndecrypt error: {err[:200]}" + ) + HfApi(token=token).change_discussion_status( + repo_id=repo_id, discussion_num=pr_num, new_status="closed", + repo_type="dataset", comment=msg, + ) + print( + f"[hf_poller] closed PR #{pr_num} ({bundle_id[:8]}): " + "decrypt failed, re-seal instructions posted" + ) + except Exception as e: # noqa: BLE001 — PR-close is best-effort; never block dedup + print(f"[hf_poller] WARN: could not close PR #{pr_num} ({bundle_id[:8]}): {e}") + + def download_one( bundle_id: str, repo_id: str, @@ -195,12 +237,17 @@ def download_one( git_ref: str = "main", pr_num: int | None = None, created_at: str | None = None, -) -> bool: +) -> str: """Download all files for one bundle into dest_dir//. git_ref is the revision to read from — `main` for legacy direct-commit flows, `refs/pr/N` for PR-based submissions (the default since miners aren't org members on RalphLabsAI). + + Returns a status string: "ok" (bundle materialised), "decrypt_failed" (blob + downloaded but could not be decrypted/unpacked — permanent, wrong/rotated seal + key; the PR is closed with re-seal instructions), or "unavailable" (transient — + list/download/network failure, safe to retry next epoch). """ from huggingface_hub import hf_hub_download, list_repo_files @@ -224,13 +271,14 @@ def download_one( bundle_files = [f for f in all_files if f.startswith(prefix)] except Exception as e: print(f"[hf_poller] list files failed for {bundle_id} @ {git_ref}: {e}") - return False + return "unavailable" if not bundle_files: print(f"[hf_poller] no files found for {bundle_id} @ {git_ref}") - return False + return "unavailable" cache = out / "_hf_cache" + decrypt_err: Optional[str] = None # set iff an enc blob downloaded but wouldn't decrypt/unpack enc_remote = f"{prefix}{bundle_crypto.ENC_FILENAME}" if enc_remote in bundle_files: # Encrypted submission: download the blob, decrypt with the validator @@ -250,6 +298,7 @@ def download_one( except Exception as e: print(f"[hf_poller] decrypt/unpack failed for {bundle_id}: {e}") success = 0 + decrypt_err = str(e) else: # Legacy plaintext: download each file into the bundle dir. success = 0 @@ -276,7 +325,12 @@ def download_one( if success == 0: shutil.rmtree(out) - return False + if decrypt_err is not None: + # Wrong/rotated seal key — permanent. Tell the miner how to re-seal and + # close the PR; the caller stamps it done so it stops churning every epoch. + _close_pr_wrong_key(repo_id, pr_num, token, bundle_id, decrypt_err) + return "decrypt_failed" + return "unavailable" # transient (network / mid-upload) — safe to retry # Annotate which PR this came from so the validator can merge later. if pr_num is not None: @@ -289,7 +343,7 @@ def download_one( }, indent=2, )) - return True + return "ok" def poll_hub( @@ -329,13 +383,19 @@ def poll_hub( for sub in new[:limit]: bid = sub["bundle_id"] print(f"[hf_poller] downloading {bid} from PR #{sub['pr_num']} ({sub['git_ref']})...") - if download_one(bid, repo_id, pending, token=token, - git_ref=sub["git_ref"], pr_num=sub["pr_num"], - created_at=sub.get("created_at")): + result = download_one(bid, repo_id, pending, token=token, + git_ref=sub["git_ref"], pr_num=sub["pr_num"], + created_at=sub.get("created_at")) + if result == "ok": downloaded.append(bid) processed[bid] = VALIDATOR_VERSION - else: - print(f"[hf_poller] skipped {bid} (download failed)") + elif result == "decrypt_failed": + # Permanent (wrong/rotated seal key). Stamp done so we don't re-poll + + # re-close it every epoch — the PR was closed with re-seal instructions. + processed[bid] = VALIDATOR_VERSION + print(f"[hf_poller] {bid[:8]}: decrypt failed — marked done (PR closed; re-seal to current pubkey)") + else: # "unavailable" — transient; leave un-stamped so it retries next epoch + print(f"[hf_poller] skipped {bid} (unavailable — will retry)") state["validator_version"] = VALIDATOR_VERSION state["processed"] = processed From 26033979f5cb095d32fb0ce816aef8b24f6a698c Mon Sep 17 00:00:00 2001 From: karpabot Date: Thu, 2 Jul 2026 17:06:01 +0000 Subject: [PATCH 8/9] hf_poller: hard timeout on HF fetches (anti-hang / slowloris) - SIGALRM wall-clock cap (_HF_FETCH_TIMEOUT_S, default 300s) around list_repo_files + hf_hub_download; huggingface_hub has no total timeout, so a stalled mid-stream fetch froze the whole epoch loop ~35min - timed-out fetch returns 'stalled' (transient) instead of hanging - poll_hub retries a stall, then marks it done after _MAX_STALLS (3) so one unfetchable/slow bundle can't re-stall the loop every epoch - closes a DoS vector: a miner could upload a deliberately-slow bundle - tests: timeout fires + fast passthrough (+ alarm restored) + stall counter Co-Authored-By: Claude Opus 4.8 --- tests/test_hf_poller_decrypt_feedback.py | 37 +++++++++++ validator/hf_poller.py | 80 +++++++++++++++++++++++- 2 files changed, 114 insertions(+), 3 deletions(-) diff --git a/tests/test_hf_poller_decrypt_feedback.py b/tests/test_hf_poller_decrypt_feedback.py index 47a6ea8..3358274 100644 --- a/tests/test_hf_poller_decrypt_feedback.py +++ b/tests/test_hf_poller_decrypt_feedback.py @@ -87,3 +87,40 @@ def _boom(token=None): monkeypatch.setattr(huggingface_hub, "HfApi", _boom) hp._close_pr_wrong_key("RalphLabsAI/proof-bundles", None, None, "c" * 64, "x") # no raise + + +# --- HF fetch timeout (hang / slowloris protection) --------------------------- + +def test_with_timeout_fires_on_slow_call(): + import time + # a call that runs longer than the cap raises _FetchTimeout (not a hang) + try: + hp._with_timeout(1, time.sleep, 5) + raise AssertionError("expected _FetchTimeout") + except hp._FetchTimeout: + pass + + +def test_with_timeout_passes_through_fast_call(): + assert hp._with_timeout(5, lambda: 42) == 42 + import signal + # handler + alarm are restored (no lingering alarm) + assert signal.getitimer(signal.ITIMER_REAL)[0] == 0.0 + + +def test_poll_hub_stalled_retries_then_marks_done(tmp_path, monkeypatch): + monkeypatch.setattr(hp, "_MAX_STALLS", 2) + hp._STALL_COUNTS.clear() + prs = [_pr("st", 1)] + monkeypatch.setattr(hp, "list_remote_submissions", lambda repo_id, token=None: prs) + monkeypatch.setattr(hp, "download_one", lambda bid, repo_id, dest, **kw: "stalled") + bid = prs[0]["bundle_id"] + + # 1st stall: retried (not done) + hp.poll_hub(tmp_path) + assert bid not in hp._load_state(tmp_path)["processed"] + assert hp._STALL_COUNTS[bid] == 1 + # 2nd stall reaches _MAX_STALLS: marked done so it can't re-stall forever + hp.poll_hub(tmp_path) + assert hp._load_state(tmp_path)["processed"].get(bid) == VALIDATOR_VERSION + assert bid not in hp._STALL_COUNTS diff --git a/validator/hf_poller.py b/validator/hf_poller.py index fb05bff..7b95c06 100644 --- a/validator/hf_poller.py +++ b/validator/hf_poller.py @@ -36,6 +36,48 @@ _MAX_BLOB_BYTES = int(os.environ.get("RALPH_MAX_BLOB_BYTES", str(2 << 30))) # 2 GiB _MAX_DECOMPRESSED_BYTES = int(os.environ.get("RALPH_MAX_DECOMPRESSED_BYTES", str(3 << 30))) # 3 GiB +# --- HF fetch timeout (hang / slowloris protection) --------------------------- +# huggingface_hub has no total-download timeout: a fetch that stalls mid-stream +# (the socket lingers with no data) blocks the ENTIRE epoch loop indefinitely — a +# single dead/slow bundle froze the validator for ~35min, and a miner could upload +# a deliberately-slow bundle to freeze scoring (DoS). A SIGALRM wall-clock cap +# bounds each fetch: on expiry it's treated as transient (retried), and a bundle +# that stalls _MAX_STALLS times in a row is marked done so it can't churn. The +# default is generous so large legit bundles (checkpoints) still finish. +_HF_FETCH_TIMEOUT_S = int(os.environ.get("RALPH_HF_FETCH_TIMEOUT_S", "300")) # 5 min +_MAX_STALLS = int(os.environ.get("RALPH_HF_MAX_STALLS", "3")) +_STALL_COUNTS: dict[str, int] = {} # bundle_id -> consecutive stalls (in-process) + + +class _FetchTimeout(Exception): + """A single HF fetch (list/download) exceeded _HF_FETCH_TIMEOUT_S.""" + + +def _with_timeout(seconds: int, fn, *args, **kwargs): + """Run fn under a hard SIGALRM wall-clock cap and restore the prior handler. + + Main-thread only (signals can't be armed off it); off the main thread it runs + uncapped — the service loop that calls poll_hub IS the main thread, so the cap + applies in production. A signal interrupts a blocked socket read (EINTR), which + is exactly the stalled-download case. + """ + import signal + import threading + + if seconds <= 0 or threading.current_thread() is not threading.main_thread(): + return fn(*args, **kwargs) + + def _fire(signum, frame): + raise _FetchTimeout(f"HF fetch exceeded {seconds}s") + + old = signal.signal(signal.SIGALRM, _fire) + signal.alarm(seconds) + try: + return fn(*args, **kwargs) + finally: + signal.alarm(0) + signal.signal(signal.SIGALRM, old) + def _assert_blob_safe(blob: bytes) -> None: """Reject an oversized / decompression-bomb decrypted blob before unpack. @@ -266,9 +308,15 @@ def download_one( } try: - all_files = list_repo_files(repo_id, repo_type="dataset", token=token, revision=git_ref) + all_files = _with_timeout( + _HF_FETCH_TIMEOUT_S, list_repo_files, + repo_id, repo_type="dataset", token=token, revision=git_ref, + ) prefix = f"submissions/{bundle_id}/" bundle_files = [f for f in all_files if f.startswith(prefix)] + except _FetchTimeout as e: + print(f"[hf_poller] list TIMEOUT for {bundle_id} @ {git_ref}: {e}") + return "stalled" except Exception as e: print(f"[hf_poller] list files failed for {bundle_id} @ {git_ref}: {e}") return "unavailable" @@ -279,12 +327,14 @@ def download_one( cache = out / "_hf_cache" decrypt_err: Optional[str] = None # set iff an enc blob downloaded but wouldn't decrypt/unpack + stalled = False # set iff a fetch timed out (transient — retry, never close the PR) enc_remote = f"{prefix}{bundle_crypto.ENC_FILENAME}" if enc_remote in bundle_files: # Encrypted submission: download the blob, decrypt with the validator # key, and unpack — reproduces the same dir a plaintext bundle would. try: - local = hf_hub_download( + local = _with_timeout( + _HF_FETCH_TIMEOUT_S, hf_hub_download, repo_id=repo_id, filename=enc_remote, repo_type="dataset", local_dir=str(cache), token=token, revision=git_ref, ) @@ -295,6 +345,10 @@ def download_one( _assert_blob_safe(blob) # DoS: cap decrypted + decompressed size before unpack bundle_crypto.unpack_bundle(blob, out) success = 1 + except _FetchTimeout as e: + print(f"[hf_poller] download TIMEOUT for {bundle_id}: {e}") + success = 0 + stalled = True except Exception as e: print(f"[hf_poller] decrypt/unpack failed for {bundle_id}: {e}") success = 0 @@ -305,7 +359,8 @@ def download_one( for remote_path in bundle_files: filename = remote_path.split("/")[-1] try: - local = hf_hub_download( + local = _with_timeout( + _HF_FETCH_TIMEOUT_S, hf_hub_download, repo_id=repo_id, filename=remote_path, repo_type="dataset", @@ -316,6 +371,10 @@ def download_one( dest = (training_dir / filename) if filename in training_files else (out / filename) shutil.copy2(local, dest) success += 1 + except _FetchTimeout as e: + print(f"[hf_poller] download TIMEOUT for {bundle_id} ({filename}): {e}") + stalled = True + break except Exception as e: print(f"[hf_poller] download {filename} failed: {e}") @@ -330,6 +389,8 @@ def download_one( # close the PR; the caller stamps it done so it stops churning every epoch. _close_pr_wrong_key(repo_id, pr_num, token, bundle_id, decrypt_err) return "decrypt_failed" + if stalled: + return "stalled" # fetch timed out — transient; caller counts + eventually skips return "unavailable" # transient (network / mid-upload) — safe to retry # Annotate which PR this came from so the validator can merge later. @@ -389,11 +450,24 @@ def poll_hub( if result == "ok": downloaded.append(bid) processed[bid] = VALIDATOR_VERSION + _STALL_COUNTS.pop(bid, None) elif result == "decrypt_failed": # Permanent (wrong/rotated seal key). Stamp done so we don't re-poll + # re-close it every epoch — the PR was closed with re-seal instructions. processed[bid] = VALIDATOR_VERSION + _STALL_COUNTS.pop(bid, None) print(f"[hf_poller] {bid[:8]}: decrypt failed — marked done (PR closed; re-seal to current pubkey)") + elif result == "stalled": + # Fetch timed out (the cap already fired, so no hang). Retry a few times + # in case it's a transient network blip; if it keeps stalling, mark it + # done so one unfetchable bundle can't re-stall the loop every epoch. + _STALL_COUNTS[bid] = _STALL_COUNTS.get(bid, 0) + 1 + if _STALL_COUNTS[bid] >= _MAX_STALLS: + processed[bid] = VALIDATOR_VERSION + _STALL_COUNTS.pop(bid, None) + print(f"[hf_poller] {bid[:8]}: download stalled {_MAX_STALLS}x — marked done (unfetchable, skipping)") + else: + print(f"[hf_poller] {bid[:8]}: download stalled ({_STALL_COUNTS[bid]}/{_MAX_STALLS}) — will retry") else: # "unavailable" — transient; leave un-stamped so it retries next epoch print(f"[hf_poller] skipped {bid} (unavailable — will retry)") From b150bed4be5378d23fef39215ae07e0e878ee45d Mon Sep 17 00:00:00 2001 From: karpabot Date: Fri, 3 Jul 2026 04:36:21 +0000 Subject: [PATCH 9/9] service: harden benchmark self-test + make idle-audit testable (#91 gaps) - extract _benchmark_selftest: now REFUSES startup (SystemExit) when active_benchmark.json is blind-forgeable AND RALPH_BENCHMARK_CROWN=1 (was warn-only); default crown-off path still just warns - extract _run_idle_audit: same RALPH_AUDIT_LOOP_OFF gate + run_pending_audits call, now a unit-testable helper instead of inline in main() - tests: refuse-on-crown+forgeable, warn-when-crown-off, missing-file noop; audit gate on/off + error-swallow Co-Authored-By: Claude Opus 4.8 --- tests/test_service_selftest_audit.py | 83 ++++++++++++++++++++++ validator/service.py | 101 ++++++++++++++++++--------- 2 files changed, 150 insertions(+), 34 deletions(-) create mode 100644 tests/test_service_selftest_audit.py diff --git a/tests/test_service_selftest_audit.py b/tests/test_service_selftest_audit.py new file mode 100644 index 0000000..d53db53 --- /dev/null +++ b/tests/test_service_selftest_audit.py @@ -0,0 +1,83 @@ +"""PR #91 gap-closing: startup benchmark self-test hard-fail + idle-epoch audit gate. + +Both behaviors were previously inline in service.main() (untestable / warn-only). +They are now extracted into _benchmark_selftest and _run_idle_audit and covered here. +""" +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +import numpy as np # noqa: E402 +import pytest # noqa: E402 + +import ralph_bootstrap # noqa: F401,E402 +from validator import service as S # noqa: E402 + + +def _forgeable(n=300): + # target systematically small, distractors uniform-large -> a blind score=-id + # sweep beats chance (the exact deployed break). + rng = np.random.default_rng(0) + return [ + { + "context_ids": [1, 2, 3], + "target_id": int(rng.integers(0, 5000)), + "distractors": [int(rng.integers(20000, 50000)) for _ in range(4)], + } + for _ in range(n) + ] + + +def _write_bench(root: Path, examples): + d = root / "eval" / "private" + d.mkdir(parents=True, exist_ok=True) + (d / "active_benchmark.json").write_text(json.dumps(examples)) + + +# --- _benchmark_selftest: warns by default, REFUSES when crown enabled + forgeable --- + +def test_selftest_refuses_when_crown_enabled_and_forgeable(tmp_path, monkeypatch): + _write_bench(tmp_path, _forgeable()) + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") + with pytest.raises(SystemExit): + S._benchmark_selftest(tmp_path) + + +def test_selftest_forgeable_but_crown_off_only_warns(tmp_path, monkeypatch): + _write_bench(tmp_path, _forgeable()) + monkeypatch.delenv("RALPH_BENCHMARK_CROWN", raising=False) + S._benchmark_selftest(tmp_path) # neutralized default -> warn, no raise + + +def test_selftest_missing_file_is_noop(tmp_path, monkeypatch): + monkeypatch.setenv("RALPH_BENCHMARK_CROWN", "1") + S._benchmark_selftest(tmp_path) # no active_benchmark.json -> return, no raise + + +# --- _run_idle_audit: default-on gate, RALPH_AUDIT_LOOP_OFF=1 disables, errors swallowed --- + +def test_idle_audit_gate_calls_then_off(tmp_path, monkeypatch): + calls = [] + import validator.audit_scheduler as sched + monkeypatch.setattr( + sched, "run_pending_audits", + lambda *a, **k: (calls.append(1), {"processed": 0})[1], + ) + monkeypatch.delenv("RALPH_AUDIT_LOOP_OFF", raising=False) + S._run_idle_audit(object(), tmp_path, 0.013) + assert len(calls) == 1 # gate default-on -> called + monkeypatch.setenv("RALPH_AUDIT_LOOP_OFF", "1") + S._run_idle_audit(object(), tmp_path, 0.013) + assert len(calls) == 1 # gate off -> NOT called again + + +def test_idle_audit_swallows_errors(tmp_path, monkeypatch): + import validator.audit_scheduler as sched + + def _boom(*a, **k): + raise RuntimeError("audit blew up") + + monkeypatch.setattr(sched, "run_pending_audits", _boom) + monkeypatch.delenv("RALPH_AUDIT_LOOP_OFF", raising=False) + assert S._run_idle_audit(object(), tmp_path, 0.013) is None # never raises diff --git a/validator/service.py b/validator/service.py index 652d98d..513b00e 100644 --- a/validator/service.py +++ b/validator/service.py @@ -1427,6 +1427,66 @@ def _generate_audit_report( ) +def _benchmark_selftest(ralph_root: Path) -> None: + """Startup blind-forgeability check for the deployed benchmark (anti-gaming P0). + + If active_benchmark.json is content-forgeable (a blind score=+/-token_id sweep + beats chance) AND the benchmark crown is enabled (RALPH_BENCHMARK_CROWN=1), + REFUSE to start — running a gameable crown is worse than not starting; the + operator must regenerate a content-whitened benchmark first. If the crown is off + (the default), the scoring neutralization already protects the crown, so just + warn. A self-test *error* (bad file / missing dep) never blocks startup. + """ + try: + from eval.benchmark import benchmark_blind_forgeable + + bp = ralph_root / "eval" / "private" / "active_benchmark.json" + if not bp.exists(): + return + forge, why = benchmark_blind_forgeable(json.loads(bp.read_text())) + if forge and os.environ.get("RALPH_BENCHMARK_CROWN") == "1": + raise SystemExit( + f"REFUSING TO START: benchmark self-test failed — {why}. " + "RALPH_BENCHMARK_CROWN=1 with a forgeable active_benchmark.json is " + "exploitable; regenerate a content-whitened benchmark or unset " + "RALPH_BENCHMARK_CROWN." + ) + (log_warn if forge else log_info)(f"benchmark self-test: {why}") + except SystemExit: + raise + except Exception as e: # noqa: BLE001 — a self-test *error* must never block startup + log_warn(f"benchmark self-test skipped: {e}") + + +def _run_idle_audit(chain, ralph_root: Path, noise_floor_margin: float): + """Drain ONE pending audit job on an idle epoch (no submissions scored). + + Was dead code (run_pending_audits called nowhere -> 0 fraud/blacklist events). + On a re-derivation mismatch run_pending_audits appends submission_fraud + blacklists + the miner. Off: RALPH_AUDIT_LOOP_OFF=1 (default on). NOTE the audit is a proxy re-run + until canonical-data / checkpoint-bound re-derivation land (plan P2/P4). Returns the + run_pending_audits result dict, or None if disabled/errored. + """ + if os.environ.get("RALPH_AUDIT_LOOP_OFF") == "1": + return None + try: + from validator.audit_scheduler import run_pending_audits + + a = run_pending_audits( + chain, ralph_root, ralph_root / "chain", + noise_floor_margin=noise_floor_margin, limit=1, + ) + if a.get("processed"): + log_info( + f"audit: processed {a['processed']} " + f"(passed {a.get('passed', 0)}, failed {a.get('failed', 0)})" + ) + return a + except Exception as e: # noqa: BLE001 — audit must never crash the epoch loop + log_warn(f"audit loop error: {e}") + return None + + def main(): p = argparse.ArgumentParser(description="Ralph continuous validator service") p.add_argument("--queue-dir", type=Path, default=RALPH_ROOT / "queue") @@ -1485,19 +1545,10 @@ def main(): else: log_info("HF Hub poll: disabled") log_info(f"submit bundles to: {args.queue_dir / 'pending' / '/'}") - # Benchmark blind-forgeability self-test (anti-gaming P0): warn loudly if the - # deployed active_benchmark.json is content-forgeable (a blind score=±token_id - # sweep beats chance), so it can never silently gate the crown once the - # benchmark is re-enabled host-reduced. - try: - from eval.benchmark import benchmark_blind_forgeable - - _bp = RALPH_ROOT / "eval" / "private" / "active_benchmark.json" - if _bp.exists(): - _forge, _why = benchmark_blind_forgeable(json.loads(_bp.read_text())) - (log_warn if _forge else log_info)(f"benchmark self-test: {_why}") - except Exception as _e: # noqa: BLE001 — self-test must never block startup - log_warn(f"benchmark self-test skipped: {_e}") + # Benchmark blind-forgeability self-test (anti-gaming P0). Warns if the deployed + # active_benchmark.json is content-forgeable; REFUSES to start if it's forgeable + # AND the crown is enabled (RALPH_BENCHMARK_CROWN=1) — see _benchmark_selftest. + _benchmark_selftest(RALPH_ROOT) log_info("") epoch = 0 @@ -1523,27 +1574,9 @@ def main(): f"{result['accepted']} accepted, {result['rejected']} rejected{mf_str}") else: log_info(f"epoch {epoch}: no pending submissions") - # Idle: drain ONE pending audit job. This was dead — run_pending_audits - # was called nowhere, so 0 fraud/blacklist events ever fired and every - # economic-deterrence claim was moot. On a re-derivation mismatch it - # appends submission_fraud + blacklists the miner (zeroed next set_weights). - # Off: RALPH_AUDIT_LOOP_OFF=1. NOTE the audit is a proxy re-run until - # canonical data + checkpoint-bound re-derivation land (plan P2/P4). - if os.environ.get("RALPH_AUDIT_LOOP_OFF") != "1": - try: - from validator.audit_scheduler import run_pending_audits - - _a = run_pending_audits( - chain, RALPH_ROOT, RALPH_ROOT / "chain", - noise_floor_margin=args.noise_floor, limit=1, - ) - if _a.get("processed"): - log_info( - f"audit: processed {_a['processed']} " - f"(passed {_a.get('passed', 0)}, failed {_a.get('failed', 0)})" - ) - except Exception as _e: # noqa: BLE001 — audit must never crash the epoch loop - log_warn(f"audit loop error: {_e}") + # Idle: drain ONE pending audit job (was dead code -> 0 fraud events). + # See _run_idle_audit; off via RALPH_AUDIT_LOOP_OFF=1. + _run_idle_audit(chain, RALPH_ROOT, args.noise_floor) except Exception as e: log_err(f"epoch {epoch} failed: {e}") log_debug(traceback.format_exc())