integrity: reject fabricated training_log (forged-down wall_clock)

tests/test_training_log_integrity.py

@@ -0,0 +1,90 @@
+"""check_training_log_integrity — reject a fabricated training_log used to disguise a
+forged-down wall_clock_s, without false-positiving on a legit low-jitter run.
+
+Root incident: taohunter v0.3.3 reported wall_clock_s low enough to hit exactly 50%
+MFU (under the 70% ceiling) and shipped a perfectly-linear log (step-0 elapsed 0.0)
+that agreed with it, keeping normalized-H100h under the cap.
+"""
+import random
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+import ralph_bootstrap  # noqa: F401,E402
+from validator.integrity import check_training_log_integrity  # noqa: E402
+
+TOK = 50 * 5243  # tokens per logged 50-step interval
+
+
+def _row(step, elapsed, tokens_seen, tps):
+    return {"step": step, "loss": 3.0, "tokens_seen": tokens_seen,
+            "tokens_per_sec": tps, "elapsed_s": elapsed}
+
+
+def _real(n=120, step=50, dt=0.808, step0=6.5, jitter=0.02, seed=1):
+    rng = random.Random(seed)
+    rows, e = [], 0.0
+    for i in range(n):
+        e = step0 if i == 0 else e + step * dt * (1 + rng.uniform(-jitter, jitter))
+        ts = (i + 1) * TOK
+        rows.append(_row(i * step, e, ts, ts / e))  # tps == tokens/elapsed identity
+    return rows
+
+
+def test_real_jittered_log_passes():
+    rows = _real()
+    ok, reason, _ = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"]})
+    assert ok, reason
+
+
+def test_step0_zero_rejected():
+    # taohunter-style: perfectly linear, step-0 elapsed 0.0
+    rows = [_row(i * 50, (i * 50) * 0.808, (i + 1) * TOK, 324500.0) for i in range(120)]
+    ok, reason, _ = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"]})
+    assert not ok and "step-0" in reason
+
+
+def test_tps_identity_evasion_rejected():
+    # sloppy fabrication: elapsed and tps produced inconsistently -> the MAJORITY of
+    # rows break the tokens_seen/elapsed_s identity (a uniform offset instead reads as
+    # a non-canonical tps definition and is warn-only, tested separately).
+    rows = _real()
+    for i, r in enumerate(rows):
+        if i % 3 != 0 and r["elapsed_s"] > 0:  # ~2/3 of rows inconsistent
+            r["tokens_per_sec"] = r["tokens_seen"] / r["elapsed_s"] * 1.5
+    ok, reason, _ = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"]})
+    assert not ok and "tokens_per_sec" in reason
+
+
+def test_wall_mismatch_rejected():
+    rows = _real()
+    ok, reason, _ = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"] * 0.5})
+    assert not ok and "wall_clock_s" in reason
+
+
+def test_perfectly_linear_but_valid_step0_and_identity_warns_not_rejects():
+    # FP-safety: a perfectly-linear log with a REAL step-0 + valid tps identity is NOT
+    # rejected (a legit stable torch.compile run can look like this) — only warned/audited.
+    rows = []
+    for i in range(120):
+        e = 6.5 + (i * 50) * 0.808  # linear after a real step-0
+        ts = (i + 1) * TOK
+        rows.append(_row(i * 50, e, ts, ts / e))
+    ok, reason, warns = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"]})
+    assert ok, reason
+    assert any("perfectly linear" in w for w in warns)  # flagged for audit
+
+
+def test_too_few_rows_deferred():
+    ok, _, _ = check_training_log_integrity([{"step": 0, "elapsed_s": 5.0}], {})
+    assert ok  # <2 rows -> deferred (op3 handles empty)
+
+
+def test_noncanonical_tps_semantics_warns_not_rejects():
+    # a recipe that logs INSTANTANEOUS tps (never == cumulative) must not be rejected
+    rows = _real()
+    for r in rows:
+        r["tokens_per_sec"] = 999999.0  # matches nothing
+    ok, reason, warns = check_training_log_integrity(rows, {"wall_clock_s": rows[-1]["elapsed_s"]})
+    assert ok, reason  # identity skipped (no row matches) rather than rejected
+    assert any("non-canonical tps" in w for w in warns)

validator/integrity.py

@@ -151,6 +151,122 @@ def check_compute_plausibility(
     return True, f"compute plausible: {tokens / wall:,.0f} tok/s, {mfu * 100:.0f}% MFU"
 
 
+# --- Training-log integrity (anti fabricated-log / forged-down wall_clock) ------
+#
+# check_compute_plausibility + the compute-budget cap both read
+# final_state.wall_clock_s, but wall_clock_s is miner-authored and never derived
+# from a measured quantity — a forger reports it LOW so a memorization-grade
+# over-budget run lands under both the normalized-H100h cap and the 70% MFU
+# ceiling, then fabricates a training log that agrees with it (observed: elapsed_s
+# exactly linear in step, step-0 elapsed 0.0 at full throughput). This guard
+# cross-checks the log against itself + wall_clock_s using only ARITHMETIC
+# IDENTITIES and a PHYSICAL floor, so it never false-positives on a legit low-jitter
+# (torch.compile) run the way a variance/linearity threshold would.
+DEFAULT_STEP0_FLOOR_S = 1.0
+
+
+def check_training_log_integrity(
+    log_rows: list[dict],
+    final_state: dict,
+    *,
+    step0_floor_s: float = DEFAULT_STEP0_FLOOR_S,
+    tps_tol: float = 0.01,
+    wall_tol_frac: float = 0.02,
+) -> tuple[bool, str, list[str]]:
+    """Reject a fabricated training_log used to disguise a forged-down wall_clock_s.
+
+    Trusts only identities/impossibilities (FP-safe), never a variance test:
+      (1) STEP-0 FLOOR: the first logged step's cumulative elapsed_s must exceed
+          `step0_floor_s` — a real first step pays CUDA-init/compile overhead
+          (legit ~6-8s); a back-computed linear log reports ~0s. Load-bearing.
+      (2) tokens_per_sec IDENTITY: the canonical trainer logs
+          tokens_per_sec == tokens_seen/elapsed_s (cumulative). If the MAJORITY of
+          rows break it beyond `tps_tol`, elapsed and throughput were produced by
+          different processes (fabrication). If NO row satisfies it the recipe
+          redefined tps semantics -> warn + skip, don't reject.
+      (3) WALL CONSISTENCY: the final row's elapsed_s must equal
+          final_state.wall_clock_s within `wall_tol_frac` (same clock).
+
+    A perfectly-linear (machine-generated) elapsed series is returned as a WARNING
+    only (audit trigger) — a legit very-stable run can have near-zero jitter.
+
+    Returns (ok, reason, warnings). ok=False -> reject at op1. NOTE: closes the
+    fabricated-log vector; wall_clock_s stays fundamentally unverifiable without
+    checkpoint-bound re-derivation on canonical data (plan P4).
+    """
+    warnings: list[str] = []
+    rows = [r for r in (log_rows or []) if isinstance(r, dict)]
+    if len(rows) < 2:
+        return True, "log-integrity: too few log rows (deferred)", warnings
+
+    # (1) step-0 compile/init floor — the load-bearing check.
+    e0 = rows[0].get("elapsed_s")
+    if isinstance(e0, (int, float)) and math.isfinite(e0) and e0 < step0_floor_s:
+        return (
+            False,
+            f"fabricated log: step-0 elapsed {e0:.3f}s < {step0_floor_s:.1f}s floor "
+            "(a real first step pays CUDA-init/compile; a back-computed linear log reports ~0)",
+            warnings,
+        )
+
+    # (2) tokens_per_sec == tokens_seen/elapsed_s identity (cumulative).
+    checked = ok_id = bad_id = 0
+    for r in rows:
+        e, ts, tps = r.get("elapsed_s"), r.get("tokens_seen"), r.get("tokens_per_sec")
+        if not all(isinstance(x, (int, float)) and math.isfinite(x) for x in (e, ts, tps)):
+            continue
+        if e <= 0 or ts <= 0 or tps <= 0:
+            continue
+        implied = ts / e
+        checked += 1
+        if abs(tps - implied) / implied <= tps_tol:
+            ok_id += 1
+        else:
+            bad_id += 1
+    if checked >= 4:
+        if ok_id == 0:
+            warnings.append(
+                "log-integrity: tokens_per_sec matches tokens_seen/elapsed_s in NO row "
+                "(non-canonical tps semantics?) — identity check skipped"
+            )
+        elif bad_id > checked // 2:
+            return (
+                False,
+                f"fabricated log: tokens_per_sec disagrees with tokens_seen/elapsed_s in "
+                f"{bad_id}/{checked} rows (elapsed and throughput fabricated independently)",
+                warnings,
+            )
+
+    # (3) final logged elapsed_s must equal final_state.wall_clock_s (same clock).
+    last_e = rows[-1].get("elapsed_s")
+    wall = (final_state or {}).get("wall_clock_s")
+    if (isinstance(last_e, (int, float)) and math.isfinite(last_e) and last_e > 0
+            and isinstance(wall, (int, float)) and math.isfinite(wall) and wall > 0
+            and abs(last_e - wall) / wall > wall_tol_frac):
+        return (
+            False,
+            f"fabricated log: final log elapsed {last_e:.1f}s != wall_clock_s {wall:.1f}s "
+            f"(> {wall_tol_frac:.0%}) — log and reported wall are different clocks",
+            warnings,
+        )
+
+    # WARN-only: degenerate (machine-linear) per-step timing — audit trigger, not a gate.
+    deltas, prev = [], None
+    for r in rows:
+        e, s = r.get("elapsed_s"), r.get("step")
+        if isinstance(e, (int, float)) and isinstance(s, (int, float)) and math.isfinite(e):
+            if prev is not None and s - prev[0] > 0:
+                deltas.append((e - prev[1]) / (s - prev[0]))
+            prev = (s, e)
+    if len(deltas) >= 8 and len({round(d, 6) for d in deltas[:-1]}) <= 1:
+        warnings.append(
+            f"log-integrity: per-step wall-time is a single value across {len(deltas) - 1} "
+            "blocks — perfectly linear elapsed (likely machine-generated); flag for audit"
+        )
+
+    return True, "log-integrity: ok", warnings
+
+
 # --- Compute-budget cap (fair "1x H100-class" contest) ------------------------
 #
 # Cap total normalized H100-hours so the crown is a FIXED compute-budget contest

validator/validator.py

@@ -51,6 +51,7 @@
     check_compute_plausibility,
     check_model_size,
     check_recipe_config_matches_proof,
+    check_training_log_integrity,
     check_training_timing,
 )
 
@@ -403,6 +404,24 @@ def op1_diff_and_integrity(
         ok_c, detail_c = check_compute_plausibility(final_state, calibration)
         if not ok_c:
             return False, detail_c
+        # Fabricated-log guard: reject a training_log forged to disguise a
+        # forged-down wall_clock_s (step-0 compile floor + tokens_per_sec identity
+        # + log/wall consistency). Disable: RALPH_LOG_INTEGRITY_OFF=1.
+        if os.environ.get("RALPH_LOG_INTEGRITY_OFF") != "1":
+            try:
+                _log_rows = [
+                    json.loads(_x)
+                    for _x in (proof_dir / "training" / "training_log.jsonl")
+                    .read_text().splitlines()
+                    if _x.strip()
+                ]
+            except (OSError, ValueError):
+                _log_rows = []
+            ok_l, detail_l, warns_l = check_training_log_integrity(_log_rows, final_state)
+            for _w in warns_l:
+                print(f"[op1] {_w}")
+            if not ok_l:
+                return False, detail_l
         # Compute-budget cap (fair 1x H100-class contest). HARD reject over
         # RALPH_H100H_BUDGET (default 5.0) normalized H100-hours; spoof-proof via a
         # fixed Hopper reference. Disable: RALPH_H100H_GATE_OFF=1.