Anti-abuse hardening from the Teutonic audit: OCI copy detection, CUDA self-kill, stuck-duel guillotine, and the cross-GPU noise harness

leoma/app/validator/copy_check.py

@@ -0,0 +1,138 @@
+"""Weight-for-weight copy detection, from registry metadata alone.
+
+``main._copies_a_king`` catches the crude copy: a different hotkey re-committing a
+king's *exact* digest. It misses the copy that matters more — identical **weights**
+repackaged with a changed README or tokenizer, which produces a *new* top-level
+manifest digest but the **same per-layer safetensor digests**. On a content-addressed
+registry, identical layer digests mean identical bytes, so that model is the king's
+weights wearing a disguise. Under the old exact-digest-only gate it sailed through and
+burned a **full multi-hour duel** before tying the king and losing.
+
+This module closes that hole for the cost of one manifest fetch (a few KB) — **no
+weight download**. It is the anti-abuse gate that matters most for a subnet whose
+entire thesis is that GPU time is the scarce resource. Ported from Teutonic's
+``check_model_copy`` (validator.py), adapted to Leoma's ``ModelRef``.
+
+It does two jobs at once:
+
+* **Reject** a copy committed *after* the king — plagiarism of the incumbent.
+* **crown_earlier**: displace the king with a byte-identical model that was pushed
+  to the registry *earlier*. That is the true original author, front-run by whoever
+  got crowned first; they should hold the crown, and no duel is needed because the
+  weights are provably identical to the reigning king's.
+
+The earlier-author decision is **consensus-safe** because it rests only on the
+registry's *own* observed push time (Harbor ``push_time`` / manifest
+``Last-Modified``), which every validator reads identically — never on a
+client-supplied annotation a miner could backdate. And it is **fail-safe**: if the
+weights are identical but the timestamps can't be established, it *rejects* rather
+than crowning, so the worst case is that a legitimate earlier author is turned away,
+never that a plagiarist is enthroned.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
+from typing import Optional
+
+from leoma.infra.model_store import ModelRef, fetch_oci_copy_info
+
+
+def _parse_registry_timestamp(ts: Optional[str]) -> Optional[datetime]:
+    """Parse an ISO-8601 or RFC-2822 timestamp to an aware UTC datetime, or None."""
+    if not ts:
+        return None
+    try:
+        dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+    except ValueError:
+        try:
+            dt = parsedate_to_datetime(ts)
+        except (TypeError, ValueError):
+            return None
+    if dt is None:
+        return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def check_model_copy(
+    challenger_repo: str,
+    challenger_digest: str,
+    king_repo: str,
+    king_digest: str,
+    *,
+    fetch=fetch_oci_copy_info,
+) -> Optional[dict]:
+    """Is the challenger a weight-for-weight copy of the king? What to do about it.
+
+    Returns ``None`` when the models genuinely differ **or** when the check cannot be
+    performed (fail open — a metadata hiccup must never block a valid submission).
+
+    On a copy, returns ``{"action": "reject"|"crown_earlier", "reason": str, ...}``.
+
+    ``fetch`` is injectable for testing; in production it is
+    :func:`~leoma.infra.model_store.fetch_oci_copy_info`.
+    """
+    if not king_repo or not king_digest:
+        return None
+
+    # Exact re-commit of the reigning king. No metadata fetch needed, and never a
+    # crown_earlier candidate: an identical top-level digest is the identical upload,
+    # so there is no distinct "earlier author" to promote.
+    if challenger_repo == king_repo and challenger_digest == king_digest:
+        return {
+            "action": "reject",
+            "reason": f"challenger is the reigning king verbatim ({challenger_digest[:19]}...)",
+            "challenger_committed_at": None,
+            "king_committed_at": None,
+        }
+
+    challenger_info = fetch(ModelRef(challenger_repo, challenger_digest))
+    if not challenger_info:
+        return None
+    king_info = fetch(ModelRef(king_repo, king_digest))
+    if not king_info:
+        return None
+
+    challenger_layers = challenger_info["safetensor_layers"]
+    king_layers = king_info["safetensor_layers"]
+
+    # A different layer count, or any single differing layer digest, means these are
+    # genuinely different weights. Not a copy — let it duel.
+    if not challenger_layers or len(challenger_layers) != len(king_layers):
+        return None
+    if any(king_layers.get(title) != digest for title, digest in challenger_layers.items()):
+        return None
+
+    # Every weight layer is byte-identical. Decide by registry-observed push time only.
+    n = len(challenger_layers)
+    c_ts, k_ts = challenger_info.get("committed_at"), king_info.get("committed_at")
+    c_src, k_src = challenger_info.get("timestamp_source"), king_info.get("timestamp_source")
+    c_dt, k_dt = _parse_registry_timestamp(c_ts), _parse_registry_timestamp(k_ts)
+
+    base = (f"all {n} .safetensors layers have identical OCI digests; "
+            f"challenger pushed_at={c_ts} ({c_src}), king pushed_at={k_ts} ({k_src})")
+    meta = {
+        "challenger_committed_at": c_ts, "king_committed_at": k_ts,
+        "challenger_timestamp_source": c_src, "king_timestamp_source": k_src,
+    }
+
+    # Fail-safe: never crown an "earlier" model unless BOTH timestamps came from the
+    # registry itself. Missing/unparseable => reject (turn a real author away rather
+    # than risk enthroning a plagiarist on backdated metadata).
+    if c_dt is None or k_dt is None or not c_src or not k_src:
+        return {"action": "reject",
+                "reason": f"copy of the king; registry timestamps unavailable, cannot verify authorship: {base}",
+                **meta}
+
+    if c_dt < k_dt:
+        return {"action": "crown_earlier",
+                "reason": f"identical weights, earlier registry push time ({c_ts} < {k_ts}); "
+                          f"the challenger is the original author, front-run by the king. {base}",
+                **meta}
+
+    return {"action": "reject", "reason": f"copy of the king, not earlier than it: {base}", **meta}
+
+
+__all__ = ["check_model_copy"]

leoma/app/validator/main.py

@@ -58,6 +58,7 @@
 )
 from leoma.app.validator import rate_limit as RL
 from leoma.app.validator.prescreen import prescreen
+from leoma.app.validator.copy_check import check_model_copy
 from leoma.app.validator.state_store import MAX_DUEL_ATTEMPTS
 from leoma.app.validator import king as K
 
@@ -69,6 +70,11 @@
 # as an operator escape hatch if the Hub's config endpoint is misbehaving.
 PRESCREEN_ENABLED = os.environ.get("LEOMA_PRESCREEN", "1") != "0"
 
+# The OCI-layer copy check + earliest-author displacement. On by default: one manifest
+# fetch catches a repackaged copy of the king that the free exact-digest check misses,
+# for the cost of a few KB instead of a multi-hour duel. Fails open if it cannot run.
+COPY_CHECK_ENABLED = os.environ.get("LEOMA_COPY_CHECK", "1") != "0"
+
 # The duel parameters (metric, n_clips, delta, alpha, bootstrap, generation knobs)
 # are NOT read from the environment any more. They are the consensus surface, and
 # they live in chain.toml as `SPEC`. An env var is per-box; a per-box exam is not
@@ -82,6 +88,11 @@
 _EVAL_CONNECT_TIMEOUT = 30.0
 _EVAL_POLL_TIMEOUT = 60.0
 
+# Validator-side backstop for a duel the eval box never finishes reporting. Generous
+# on purpose (~18h at 12s blocks): a real duel of two 14B models can run hours, so this
+# only fires on a box that is genuinely wedged, not on slow-but-alive work.
+MAX_INFLIGHT_BLOCKS = int(os.environ.get("LEOMA_MAX_INFLIGHT_BLOCKS", "5400"))
+
 
 def _seen_key(hotkey: str, digest: str) -> str:
     return f"{hotkey}|{digest}"
@@ -190,6 +201,18 @@ async def start_duel(entry: ChallengerEntry, king: dict, block_hash: str) -> str
         return resp.json()["eval_id"]
 
 
+async def cancel_duel(eval_id: str) -> None:
+    """Best-effort ``DELETE /eval/{id}`` — ask the box to stop burning GPU. Never raises."""
+    import httpx
+
+    timeout = httpx.Timeout(_EVAL_POLL_TIMEOUT, connect=_EVAL_CONNECT_TIMEOUT)
+    try:
+        async with httpx.AsyncClient(timeout=timeout) as client:
+            await client.delete(f"{EVAL_SERVER_URL}/eval/{eval_id}")
+    except Exception:  # noqa: BLE001 — the abandon must proceed whether or not this lands
+        pass
+
+
 async def poll_duel(eval_id: str) -> dict:
     """Ask the eval server how a dispatched duel is going.
 
@@ -582,7 +605,29 @@ async def settle_inflight(
 
     status = result.get("status")
     if status == "running":
-        log(f"Duel {slot['eval_id']} still running (phase={result.get('phase')})", "info")
+        # The eval server has its own forward-progress watchdog, and normally it fires
+        # first. This is the validator-side backstop for the case that watchdog can't
+        # catch: a box that keeps reporting "running" without ever tripping a phase
+        # budget (a disabled watchdog, a lying box, a partition where poll succeeds but
+        # the box is wedged). Without it, one stuck duel holds the single in-flight slot
+        # forever and no other challenger is ever dispatched. The bound is deliberately
+        # generous — a real 32-clip duel of two 14B models can legitimately run hours —
+        # so this only ever fires on a genuinely hung box.
+        age = block - int(slot.get("dispatched_block", block))
+        if age > MAX_INFLIGHT_BLOCKS:
+            log(f"Duel {slot['eval_id']} has been in flight {age} blocks (> {MAX_INFLIGHT_BLOCKS}) "
+                "— abandoning as a stuck box and re-dispatching next tick", "warn")
+            await cancel_duel(slot["eval_id"])
+            state.inflight = None
+            state.touch()
+            # TRANSIENT, never LOCAL or a strike: a hung box is not the challenger's
+            # fault. Backoff + the 4-attempt budget means a genuinely pathological model
+            # that hangs every time still quarantines, while a one-off box wedge retries.
+            failure = DuelFailure(ErrorClass.TRANSIENT, "inflight_timeout",
+                                  f"duel exceeded {MAX_INFLIGHT_BLOCKS} blocks in flight")
+            await _note_failure(state, store, uid_map, entry, key, failure, block)
+            return True
+        log(f"Duel {slot['eval_id']} still running (phase={result.get('phase')}, age={age} blocks)", "info")
         return False
 
     # Terminal, one way or another: the slot is free from here on.
@@ -659,6 +704,50 @@ async def settle_inflight(
     return True
 
 
+async def _crown_earlier(
+    subtensor: bt.AsyncSubtensor,
+    wallet: bt.Wallet,
+    state: KingState,
+    uid_map: dict[str, int],
+    store: JsonBucketStore,
+    entry: ChallengerEntry,
+    key: str,
+    copy: dict,
+    block: int,
+) -> None:
+    """Displace the king with a byte-identical model that was pushed to the registry earlier.
+
+    No duel: the weights are provably identical to the reigning king's, so there is
+    nothing to evaluate — the only question was authorship, and the registry's own push
+    time answered it. The challenger is the original, front-run by whoever got crowned
+    first, and it takes the crown as a synthetic accepted verdict.
+    """
+    log(f"{entry.hotkey[:12]}... has the king's exact weights but an EARLIER registry "
+        f"push — crowning the original author, no duel. {copy['reason']}", "warn")
+
+    verdict = {
+        "accepted": True,
+        "verdict": "crown_earlier",
+        "challenge_id": f"block-{entry.block}",
+        "reason": copy["reason"],
+        "challenger_committed_at": copy.get("challenger_committed_at"),
+        "king_committed_at": copy.get("king_committed_at"),
+        "produced_at": datetime.now(timezone.utc).isoformat(),
+    }
+    state.clear_attempts(key)
+    state.mark_seen(key)
+    RL.record_verdict(state.duels, entry.hotkey, king=state.king, block=block)
+    state.record_duel(_duel_history_entry(entry, verdict, uid_map))
+    state.king, state.king_chain = K.crown(
+        state.king, state.king_chain, hotkey=entry.hotkey, model_repo=entry.model_repo,
+        model_digest=entry.model_digest, block=entry.block, challenge_id=verdict["challenge_id"],
+    )
+    state.stats["accepted"] = state.stats.get("accepted", 0) + 1
+    state.touch()
+    await state.flush(store)
+    await maybe_set_weights(subtensor, wallet, state, uid_map, store, force=True)
+
+
 async def process_challengers(
     subtensor: bt.AsyncSubtensor,
     wallet: bt.Wallet,
@@ -724,6 +813,29 @@ async def process_challengers(
             await _note_failure(state, store, uid_map, entry, key, failure, block)
             continue
 
+        # The exact-digest check above misses a copy that changed only its README or
+        # tokenizer: identical WEIGHTS, new top-level digest. This catches it from OCI
+        # layer digests — one manifest fetch, no weight download — and also displaces
+        # the king when the challenger is the byte-identical ORIGINAL, front-run by
+        # whoever got crowned first. Both decisions rest only on registry-observed
+        # push time, so validators agree; a metadata hiccup fails OPEN (returns None).
+        if COPY_CHECK_ENABLED and state.king:
+            try:
+                copy = await asyncio.to_thread(
+                    check_model_copy, entry.model_repo, entry.model_digest,
+                    state.king.get("model_repo", ""), state.king.get("model_digest", ""),
+                )
+            except Exception:  # noqa: BLE001 — the check itself must never crash the tick
+                copy = None
+            if copy and copy["action"] == "reject":
+                failure = DuelFailure(ErrorClass.PERMANENT, "copy_of_king", copy["reason"])
+                RL.record_strike(state.duels, entry.hotkey, failure.reason)
+                await _note_failure(state, store, uid_map, entry, key, failure, block)
+                continue
+            if copy and copy["action"] == "crown_earlier":
+                await _crown_earlier(subtensor, wallet, state, uid_map, store, entry, key, copy, block)
+                return  # the king changed with no duel; re-scan next tick
+
         # The GPU is the scarcest thing in the subnet. A hotkey that has just been
         # dueled, or has already spent its allowance against this king, waits.
         limited = RL.check(state.duels, entry.hotkey, king=state.king, block=block)

leoma/app/validator/rate_limit.py

@@ -7,9 +7,15 @@
 only GPU in the subnet. Nothing stops one miner from starving everyone else, and it
 costs them nothing but an upload.
 
-The reference subnet's answer (burn the miner's slot) does not transfer: their eval
-is cheap. Ours costs *hours*, so the limiter is tuned to the thing that is actually
-scarce here — GPU time, not slots.
+The reference subnet's answer is a hard 1-hotkey-1-eval burn at enqueue: a hotkey gets
+exactly one evaluation, ever. That is strictly *more* spam-proof than what is here, and
+it is independent of eval cost — so "their eval is cheap and ours isn't" is not the
+reason to prefer the softer gate. The real reason is a deliberate product choice: Leoma
+keys its seen-set on ``hotkey|digest`` so a miner can **iterate** — fix a model and
+resubmit under the same hotkey — which a permanent burn would forbid. The cost of that
+choice is that a hotkey can mint fresh digests, so this limiter (cooldown + per-reign
+cap) is what re-imposes a *cost* gate on top of the *idempotency* gate, tuned to the
+thing that is actually scarce here: GPU time.
 
 Three rules, all pure functions of state the validator already has: