diff --git a/leoma/app/validator/copy_check.py b/leoma/app/validator/copy_check.py
new file mode 100644
index 0000000..86c7fb1
--- /dev/null
+++ b/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"]
diff --git a/leoma/app/validator/main.py b/leoma/app/validator/main.py
index 4fe72bf..c80fdbb 100644
--- a/leoma/app/validator/main.py
+++ b/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)
diff --git a/leoma/app/validator/rate_limit.py b/leoma/app/validator/rate_limit.py
index c53bb23..7beda0f 100644
--- a/leoma/app/validator/rate_limit.py
+++ b/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:
 
diff --git a/leoma/delivery/commands.py b/leoma/delivery/commands.py
index 58e8d83..ba06b49 100644
--- a/leoma/delivery/commands.py
+++ b/leoma/delivery/commands.py
@@ -241,6 +241,107 @@ def corpus_verify(sample):
     log(f"{checked} clips decoded byte-identically to the manifest — this box can duel", "success")
 
 
+@cli.group()
+def calibrate():
+    """Measure the cross-GPU noise floor so delta_threshold can be set, not guessed.
+
+    Leoma crowns a challenger whose bootstrap LCB beats the king by delta_threshold.
+    That margin only means anything if it is wider than the hardware noise: the same
+    model generates slightly different frames on different GPUs, hence different
+    distances, hence potentially a different verdict. This is the subnet's largest open
+    consensus risk. Two steps:
+
+      1. `leoma calibrate generate` on EACH GPU type in the fleet -> one record per box.
+      2. `leoma calibrate analyze box_*.json` -> the measured floor + a delta recommendation.
+    """
+
+
+@calibrate.command("generate")
+@click.option("--model", default="", help="repo@digest to calibrate on (default: the pinned seed king)")
+@click.option("--n-clips", type=int, default=32, help="How many clips to generate")
+@click.option("--gpu", "gpu_name", default="", help="Label for this box (default: the detected GPU name)")
+@click.option("--out", "-o", default="calibration.json", help="Where to write this box's record")
+def calibrate_generate(model, n_clips, gpu_name, out):
+    """Generate the calibration model on this box and record per-clip distances.
+
+    Run this ONCE ON EACH GPU TYPE in the fleet. Every box uses the same model, clips
+    and seed, so any distance difference between two boxes is pure hardware noise.
+    """
+    import json
+
+    from leoma.bootstrap import emit_log as log, emit_header as log_header
+    from leoma.eval.calibrate import box_record
+    from leoma.eval.determinism import runtime_env
+    from leoma.infra.chain_config import SEED_REPO, SEED_DIGEST, SPEC
+    from leoma.infra.model_store import ModelRef
+
+    log_header("Calibration — generate")
+    SPEC.require_duel_ready()
+
+    if model:
+        repo, _, digest = model.partition("@")
+        ref = ModelRef(repo, digest)
+    elif SEED_DIGEST:
+        ref = ModelRef(SEED_REPO, SEED_DIGEST)
+    else:
+        raise click.ClickException(
+            "no --model given and chain.toml [seed].seed_digest is not pinned; "
+            "pass --model repo@digest to calibrate on a specific model"
+        )
+
+    label = gpu_name or (runtime_env().get("gpu") or "unknown-gpu")
+    log(f"Model: {ref.immutable_ref}  GPU: {label}  clips: {n_clips}", "info")
+    log("Generating (this uses the real duel path — expect it to take a while)...", "info")
+
+    record = box_record(ref, spec=SPEC, n_clips=n_clips, gpu_name=label)
+    with open(out, "w") as f:
+        json.dump(record, f, indent=2)
+
+    log_header("Done")
+    log(f"Wrote {len(record['clips'])} clip distances for {label} -> {out}", "success")
+    log("Run this on every GPU type, then: leoma calibrate analyze <all the records>", "info")
+
+
+@calibrate.command("analyze")
+@click.argument("records", nargs=-1, type=click.Path(exists=True), required=True)
+@click.option("--safety", type=float, default=3.0, help="Multiple of the noise floor to recommend")
+def calibrate_analyze(records, safety):
+    """Compare box records and print the measured floor + a delta_threshold recommendation.
+
+    Pure analysis — no GPU. Give it the JSON records from `generate` on each box.
+    """
+    import json
+
+    from leoma.bootstrap import emit_log as log, emit_header as log_header
+    from leoma.eval.calibrate import analyze
+    from leoma.infra.chain_config import SPEC
+
+    log_header("Calibration — analyze")
+    loaded = []
+    for path in records:
+        with open(path) as f:
+            loaded.append(json.load(f))
+        log(f"loaded {path}: {loaded[-1].get('gpu')} ({len(loaded[-1].get('clips', []))} clips)", "info")
+
+    report = analyze(loaded, current_delta_threshold=SPEC.duel.delta_threshold, safety_factor=safety)
+
+    log_header("Cross-GPU noise")
+    log(f"boxes: {report.n_boxes}  pairs: {report.n_pairs}  gpus: {', '.join(report.gpus)}", "info")
+    for p in report.pairs:
+        repro = "bit-identical" if p.bit_identical_clips == p.n_clips else \
+                f"{p.bit_identical_clips}/{p.n_clips} clips identical"
+        log(f"  {p.gpu_a} vs {p.gpu_b}: |Δ| max={p.abs_delta_max:.2e} p99={p.abs_delta_p99:.2e}  "
+            f"mu_hat={p.mu_hat:+.2e} lcb={p.lcb:+.2e} ucb={p.ucb:+.2e}  ({repro})", "info")
+
+    log(f"worst spurious mean-advantage (any pair): {report.max_abs_mu_hat:.3e}", "info")
+    log(f"worst single-clip disagreement:           {report.max_abs_delta:.3e}", "info")
+    log(f"recommended delta_threshold (>= {safety}x): {report.recommended_delta_threshold}", "success")
+
+    level = "success" if report.verdict.startswith("PASS") else \
+            "error" if report.verdict.startswith("FAIL") else "info"
+    log(report.verdict, level)
+
+
 @cli.group()
 def miner():
     """Miner management commands.
diff --git a/leoma/eval/calibrate.py b/leoma/eval/calibrate.py
new file mode 100644
index 0000000..fd6f0a4
--- /dev/null
+++ b/leoma/eval/calibrate.py
@@ -0,0 +1,268 @@
+"""Measure the cross-GPU noise floor, so ``delta_threshold`` can be set, not guessed.
+
+This is the tool for the subnet's single largest open consensus risk. Leoma crowns a
+challenger when its bootstrap lower-confidence-bound beats the king by
+``delta_threshold``. That margin only means anything if it is *wider than the noise*:
+the same model, generating the same clips from the same seed, produces slightly
+different frames on an H100 than on an A100 (14B bf16 diffusion is not bit-exact across
+architectures), hence slightly different distances, hence a slightly different verdict.
+If two honest validators land on opposite sides of the margin on a near-threshold
+challenger, they fork the king chain forever. Today ``delta_threshold`` is a hopeful
+constant; this makes it a *measured multiple of a known floor*.
+
+**The experiment.** Generate one real model on the calibration clips on *each* GPU
+type in the fleet (:func:`box_record`, run on the box). Then, for every pair of boxes,
+the per-clip distance difference ``d_i = dist_A[i] - dist_B[i]`` is exactly the
+cross-GPU noise — because the model, the clip, the seed and the truth are all
+identical, so *only the hardware differs*. A self-duel between two honest boxes'
+generations of the same model has a true mean advantage of **zero**; whatever nonzero
+``mean(d)`` the bootstrap finds is pure hardware noise, and its magnitude is how far a
+marginal verdict can stray from the truth. :func:`analyze` computes that across all
+pairs and recommends a ``delta_threshold`` above it.
+
+The analysis here is pure and unit-tested. :func:`box_record` needs a GPU and reuses
+the production generation + scoring path, so it is exercised on real hardware, not in
+CI.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional, Sequence
+
+import numpy as np
+
+from leoma.eval.bootstrap import paired_bootstrap_verdict
+
+#: How far above the measured noise floor to set the margin. The recommendation is
+#: this times the worst spurious mean-advantage any box pair produced.
+DEFAULT_SAFETY_FACTOR = 3.0
+
+
+@dataclass(frozen=True)
+class PairResult:
+    """Cross-GPU noise between one pair of boxes, on the same model + clips."""
+
+    gpu_a: str
+    gpu_b: str
+    n_clips: int
+    bit_identical_clips: int          # clips where both boxes generated byte-identical frames
+    abs_delta_mean: float
+    abs_delta_p95: float
+    abs_delta_p99: float
+    abs_delta_max: float
+    mu_hat: float                     # mean(dist_A - dist_B); true value is 0, so this is bias
+    lcb: float
+    ucb: float
+
+    def as_dict(self) -> dict:
+        return {
+            "gpu_a": self.gpu_a, "gpu_b": self.gpu_b, "n_clips": self.n_clips,
+            "bit_identical_clips": self.bit_identical_clips,
+            "abs_delta_mean": self.abs_delta_mean, "abs_delta_p95": self.abs_delta_p95,
+            "abs_delta_p99": self.abs_delta_p99, "abs_delta_max": self.abs_delta_max,
+            "mu_hat": self.mu_hat, "lcb": self.lcb, "ucb": self.ucb,
+        }
+
+
+@dataclass(frozen=True)
+class CalibrationReport:
+    n_boxes: int
+    n_pairs: int
+    gpus: tuple[str, ...]
+    #: Worst spurious mean-advantage across all pairs — the number delta must beat.
+    max_abs_mu_hat: float
+    #: Worst single-clip distance disagreement across all pairs.
+    max_abs_delta: float
+    #: The fraction of box-pairs that were byte-identical (fully reproducible).
+    fully_reproducible_pairs: int
+    recommended_delta_threshold: float
+    current_delta_threshold: Optional[float]
+    verdict: str                      # human-readable pass/fail against the current pin
+    pairs: tuple[PairResult, ...]
+
+    def as_dict(self) -> dict:
+        return {
+            "n_boxes": self.n_boxes, "n_pairs": self.n_pairs, "gpus": list(self.gpus),
+            "max_abs_mu_hat": self.max_abs_mu_hat, "max_abs_delta": self.max_abs_delta,
+            "fully_reproducible_pairs": self.fully_reproducible_pairs,
+            "recommended_delta_threshold": self.recommended_delta_threshold,
+            "current_delta_threshold": self.current_delta_threshold,
+            "verdict": self.verdict,
+            "pairs": [p.as_dict() for p in self.pairs],
+        }
+
+
+def _clip_map(record: dict) -> dict[str, dict]:
+    return {c["clip_id"]: c for c in record.get("clips", [])}
+
+
+def _validate_records(records: Sequence[dict]) -> list[str]:
+    """Every box must have scored the SAME model on the SAME clips, or the pairwise
+    difference is measuring a different exam, not hardware noise."""
+    if len(records) < 2:
+        raise ValueError("need at least 2 box records to measure cross-GPU noise")
+
+    models = {r.get("model") for r in records}
+    if len(models) != 1:
+        raise ValueError(f"records are for different models {models}; the calibration model "
+                         "must be identical across boxes or the difference is not hardware noise")
+
+    corpora = {r.get("corpus_id") for r in records}
+    if len(corpora) != 1:
+        raise ValueError(f"records are for different corpora {corpora}")
+
+    clip_sets = [frozenset(_clip_map(r)) for r in records]
+    common = set.intersection(*(set(s) for s in clip_sets))
+    if not common:
+        raise ValueError("box records share no common clip_ids")
+    return sorted(common)
+
+
+def _pair(a: dict, b: dict, clip_ids: Sequence[str], *, alpha: float, n_bootstrap: int, seed: int) -> PairResult:
+    am, bm = _clip_map(a), _clip_map(b)
+    da = np.array([am[c]["distance"] for c in clip_ids], dtype=np.float64)
+    db = np.array([bm[c]["distance"] for c in clip_ids], dtype=np.float64)
+    delta = da - db
+    abs_delta = np.abs(delta)
+
+    bit_identical = sum(
+        1 for c in clip_ids
+        if am[c].get("frames_digest") and am[c].get("frames_digest") == bm[c].get("frames_digest")
+    )
+
+    # A self-duel between the two boxes' generations: true advantage is 0, so mu_hat is
+    # bias and [lcb, ucb] brackets how far a marginal verdict can drift. Reuse the exact
+    # production bootstrap so the number means what the duel means.
+    v = paired_bootstrap_verdict(da, db, delta_threshold=0.0, alpha=alpha,
+                                 n_bootstrap=n_bootstrap, seed=seed)
+    # The two-sided concern: symmetric bound on |mean advantage|.
+    ucb = float(np.quantile(_bootstrap_means(delta, n_bootstrap, seed), 1.0 - alpha))
+
+    return PairResult(
+        gpu_a=str(a.get("gpu", "?")), gpu_b=str(b.get("gpu", "?")),
+        n_clips=len(clip_ids), bit_identical_clips=bit_identical,
+        abs_delta_mean=round(float(abs_delta.mean()), 8),
+        abs_delta_p95=round(float(np.quantile(abs_delta, 0.95)), 8),
+        abs_delta_p99=round(float(np.quantile(abs_delta, 0.99)), 8),
+        abs_delta_max=round(float(abs_delta.max()), 8),
+        mu_hat=round(float(delta.mean()), 8), lcb=v["lcb"], ucb=round(ucb, 8),
+    )
+
+
+def _bootstrap_means(x: np.ndarray, n_bootstrap: int, seed: int) -> np.ndarray:
+    rng = np.random.default_rng(seed)
+    n = x.shape[0]
+    out = np.empty(n_bootstrap, dtype=np.float64)
+    for i in range(n_bootstrap):
+        out[i] = x[rng.integers(0, n, size=n)].mean()
+    return out
+
+
+def analyze(
+    records: Sequence[dict],
+    *,
+    current_delta_threshold: Optional[float] = None,
+    safety_factor: float = DEFAULT_SAFETY_FACTOR,
+    alpha: float = 0.001,
+    n_bootstrap: int = 10_000,
+    seed: int = 0,
+) -> CalibrationReport:
+    """Turn N per-box records into a cross-GPU noise report + a delta recommendation."""
+    clip_ids = _validate_records(records)
+
+    pairs: list[PairResult] = []
+    for i in range(len(records)):
+        for j in range(i + 1, len(records)):
+            pairs.append(_pair(records[i], records[j], clip_ids,
+                               alpha=alpha, n_bootstrap=n_bootstrap, seed=seed))
+
+    # The number delta must beat: the worst spurious mean-advantage any pair could show
+    # on two models that are actually identical. We take the symmetric bound (max of
+    # |lcb|, |ucb|) so a challenger cannot win by noise in either direction.
+    max_abs_mu = max((max(abs(p.lcb), abs(p.ucb)) for p in pairs), default=0.0)
+    max_abs_delta = max((p.abs_delta_max for p in pairs), default=0.0)
+    fully_repro = sum(1 for p in pairs if p.bit_identical_clips == p.n_clips)
+
+    recommended = round(safety_factor * max_abs_mu, 6)
+
+    if current_delta_threshold is None:
+        verdict = f"recommend delta_threshold >= {recommended} ({safety_factor}x the measured floor)"
+    elif current_delta_threshold >= recommended:
+        verdict = (f"PASS: current delta_threshold {current_delta_threshold} clears the measured "
+                   f"noise floor (recommended >= {recommended})")
+    else:
+        verdict = (f"FAIL: current delta_threshold {current_delta_threshold} is BELOW the measured "
+                   f"noise floor — two honest validators can fork. Raise it to >= {recommended}")
+
+    gpus = tuple(dict.fromkeys(str(r.get("gpu", "?")) for r in records))
+    return CalibrationReport(
+        n_boxes=len(records), n_pairs=len(pairs), gpus=gpus,
+        max_abs_mu_hat=round(max_abs_mu, 8), max_abs_delta=round(max_abs_delta, 8),
+        fully_reproducible_pairs=fully_repro,
+        recommended_delta_threshold=recommended,
+        current_delta_threshold=current_delta_threshold,
+        verdict=verdict, pairs=tuple(pairs),
+    )
+
+
+def box_record(model_ref, *, spec, n_clips: int, gpu_name: str) -> dict:
+    """Generate ``model_ref`` on ``n_clips`` calibration clips on THIS box and score each.
+
+    Runs the production generation + scoring path (so the measured noise is the real
+    duel's noise), model-vs-truth only — no opponent. Needs a GPU; not unit-tested.
+    Produces the record :func:`analyze` consumes.
+    """
+    from leoma.infra.model_store import ModelRef, materialize_model
+    from leoma.infra.storage_backend import create_source_read_client
+    from leoma.eval.video_runner import GenParams, load_video_pipeline, generate
+    from leoma.eval.dataset import build_duel_clips, fetch_manifest
+    from leoma.eval.determinism import apply_determinism, runtime_env
+    from leoma.eval.digests import digest_frames
+    from leoma.eval.metrics import get_metric
+    from leoma.app.validator.seeds import eval_seed, clip_generation_seed
+
+    apply_determinism(spec.determinism)
+    ref = model_ref if isinstance(model_ref, ModelRef) else ModelRef(*model_ref)
+    snapshot = materialize_model(ref)
+    params = GenParams.from_spec(spec.gen)
+    pipe = load_video_pipeline(snapshot, gen=spec.gen)
+
+    # A fixed calibration seed — the point is that every box uses the SAME seed, so any
+    # distance difference is hardware, not sampling.
+    master_seed = eval_seed("calibration", ref.immutable_ref, spec.duel.base_seed)
+    client = create_source_read_client()
+    manifest = fetch_manifest(client, spec.corpus)
+    clips, _ = build_duel_clips(
+        manifest, client=client, bucket=spec.corpus.bucket, master_seed=master_seed,
+        n_clips=n_clips, gen=params, prompt_mode=spec.gen.prompt_mode, fixed_prompt=spec.gen.prompt,
+    )
+    distance_fn = get_metric(spec.duel.metric, device=spec.duel.metric_device)
+
+    out_clips = []
+    for clip in clips:
+        gseed = clip_generation_seed(master_seed, clip.clip_index)
+        frames = generate(pipe, clip, gseed)
+        out_clips.append({
+            "clip_id": clip.clip_id,
+            "distance": float(distance_fn(frames, clip.truth_frames)),
+            "frames_digest": digest_frames(frames),
+        })
+
+    return {
+        "gpu": gpu_name,
+        "model": ref.immutable_ref,
+        "corpus_id": manifest.corpus_id,
+        "metric": spec.duel.metric,
+        "master_seed": master_seed,
+        "env": runtime_env(),
+        "clips": out_clips,
+    }
+
+
+__all__ = [
+    "DEFAULT_SAFETY_FACTOR",
+    "CalibrationReport",
+    "PairResult",
+    "analyze",
+    "box_record",
+]
diff --git a/leoma/eval_server.py b/leoma/eval_server.py
index 63f0d30..990eba5 100644
--- a/leoma/eval_server.py
+++ b/leoma/eval_server.py
@@ -118,6 +118,86 @@ class EvalRequest(BaseModel):
 #: SSE poll interval. The event log is a list; checking its length is free.
 STREAM_POLL_SECONDS = 0.05
 
+# ---------------------------------------------------------------------------
+# Fatal-CUDA self-kill
+# ---------------------------------------------------------------------------
+# Once a CUDA context is corrupted — an illegal memory access, a device-side assert,
+# a cuBLAS execution failure — the VRAM allocator and every stream on this process
+# are poisoned. Every subsequent `.from_pretrained` / generate / `empty_cache` will
+# keep raising against the same dead context. The watchdog's `os._exit` only fires
+# after a duel *stalls*; a duel that FAILS FAST on a CUDA fault would release the lock
+# and cheerfully hand the next challenger a poisoned box, mis-rejecting an honest model
+# as broken. For Leoma this is expensive in a way it is not for a text subnet: a
+# poisoned box wastes *hours* of the next duel, not minutes.
+#
+# The only recovery is to exit so the supervisor restarts with a fresh CUDA context.
+# We delay briefly so the in-flight error event reaches the validator (which classifies
+# `cuda_fatal` as TRANSIENT and retries against a — now freshly restarted — box), then
+# `os._exit` to skip atexit hooks that would touch the corrupted GPU and hang.
+# Ported from Teutonic's eval_server self-kill, which cites a real 2.5h box degradation.
+_CUDA_FATAL_TOKENS = (
+    "an illegal memory access",
+    "cudaerrorillegaladdress",
+    "device-side assert",
+    "cuda error: misaligned address",
+    "cuda error: unspecified launch failure",
+    "cuda error: an illegal instruction",
+    "cublas_status_execution_failed",
+    "cublas_status_not_initialized",
+    "cudnn_status_execution_failed",
+    "bus error",
+    "segmentation fault",
+)
+CUDA_FATAL_EXIT_DELAY_S = float(os.environ.get("LEOMA_CUDA_FATAL_DELAY", "3"))
+CUDA_FATAL_EXIT_CODE = int(os.environ.get("LEOMA_CUDA_FATAL_EXIT_CODE", "75"))
+_self_kill_scheduled = threading.Event()
+
+
+def is_cuda_fatal(exc_or_msg) -> bool:
+    """Does this error mean the CUDA context is unrecoverable (not just this duel)?"""
+    return any(tok in str(exc_or_msg or "").lower() for tok in _CUDA_FATAL_TOKENS)
+
+
+def schedule_self_kill(reason: str, *, delay_s: Optional[float] = None) -> None:
+    """Exit the process because CUDA state is poisoned. Idempotent; delayed.
+
+    Safe only because *the chain is the queue*: the restart loses at most the current
+    duel, which the validator re-dispatches next tick. Holding a poisoned box would
+    instead corrupt every future verdict.
+    """
+    if _self_kill_scheduled.is_set():
+        return
+    _self_kill_scheduled.set()
+    delay = CUDA_FATAL_EXIT_DELAY_S if delay_s is None else float(delay_s)
+
+    def _die():
+        try:
+            time.sleep(delay)  # let the SSE error event reach the validator first
+        except Exception:
+            pass
+        os._exit(CUDA_FATAL_EXIT_CODE)
+
+    threading.Thread(target=_die, daemon=True, name="leoma-cuda-fatal-self-kill").start()
+
+
+def _install_cuda_excepthook() -> None:
+    """Self-kill on a CUDA-fatal error that escapes a daemon thread (e.g. a diffusers
+    worker). Without this such an error prints a traceback and the process limps on
+    with a corrupted context, poisoning every later duel."""
+    prior = threading.excepthook
+
+    def _hook(args):
+        try:
+            if is_cuda_fatal(f"{args.exc_type.__name__}: {args.exc_value}"):
+                schedule_self_kill(f"uncaught in thread {getattr(args.thread, 'name', '?')}")
+        finally:
+            prior(args)
+
+    threading.excepthook = _hook
+
+
+_install_cuda_excepthook()
+
 
 @dataclass
 class _Job:
@@ -238,6 +318,12 @@ def _execute(job: _Job, req: EvalRequest) -> None:
             # the worker would otherwise leave the job with NO terminal event, the
             # stream hanging, and the lock held. Every exit is a terminal event.
             job.finish("error", {"error": str(e), "reason": getattr(e, "reason", "")})
+            # A CUDA-fatal error is reported cleanly here AND self-kills: the context
+            # is poisoned, so releasing the lock and serving the next duel would just
+            # hand it a dead GPU. The delay lets this error event reach the validator
+            # (which retries the challenger against the restarted box) before we exit.
+            if is_cuda_fatal(e):
+                schedule_self_kill(f"duel {job.eval_id}: {type(e).__name__}: {e}")
         finally:
             _release(job.eval_id)
 
diff --git a/leoma/infra/model_store.py b/leoma/infra/model_store.py
index 04e08ea..9134faf 100644
--- a/leoma/infra/model_store.py
+++ b/leoma/infra/model_store.py
@@ -442,6 +442,89 @@ def sha256_safetensors(path: str | os.PathLike[str]) -> str:
     return h.hexdigest()
 
 
+def fetch_oci_copy_info(ref: "ModelRef") -> Optional[dict]:
+    """Per-layer weight digests + the registry's own push timestamp, from OCI metadata.
+
+    This is what lets the validator catch a *repackaged* copy of the king — identical
+    weights re-uploaded with a changed README or tokenizer, which yields a new
+    top-level manifest digest but the SAME per-layer safetensor digests — for the cost
+    of one manifest fetch (a few KB), with **no weight download at all**. On a
+    content-addressed registry, identical layer digests mean identical bytes.
+
+    Returns ``{"safetensor_layers": {title: digest}, "committed_at": iso|None,
+    "timestamp_source": str|None}``, or **None** when the check cannot be performed
+    (an ``hf:`` ref, the registry is unreachable, the manifest is absent). None means
+    "don't know" — the caller must fail *open*, never blocking a valid submission on a
+    metadata hiccup.
+
+    ``committed_at`` is deliberately the **registry-observed** push time (Harbor's
+    ``push_time`` or the manifest's ``Last-Modified``), never a client-supplied
+    annotation like ``org.opencontainers.image.created`` — a miner can backdate the
+    latter to steal an earlier-author claim. Ported from Teutonic's
+    ``_fetch_model_oci_info``.
+    """
+    if ref.digest.startswith("hf:"):
+        return None
+    try:
+        import httpx
+        from hippius_hub._harbor import harbor_get_artifact, split_repo_id
+        from hippius_hub._oci import manifest_url, oci_headers
+        from hippius_hub.auth import (
+            get_oci_bearer_token,
+            resolve_auth_header,
+            resolve_token_value,
+        )
+        from hippius_hub.constants import resolve_registry
+        from hippius_hub.file_download import _oci_repo_path
+
+        registry = resolve_registry(None)
+        oci_repo = _oci_repo_path(ref.repo, None)
+        raw_token = _resolve_hub_token(f"copy-check manifest {ref.repo}")
+        oci_token = get_oci_bearer_token(oci_repo, resolve_token_value(raw_token), push=False)
+
+        resp = httpx.get(
+            manifest_url(registry, oci_repo, ref.digest),
+            headers=oci_headers(oci_token),
+            timeout=httpx.Timeout(15.0),
+        )
+        if resp.status_code == 404:
+            return None
+        resp.raise_for_status()
+        manifest = resp.json()
+
+        safetensor_layers: dict[str, str] = {}
+        for layer in manifest.get("layers", []):
+            title = layer.get("annotations", {}).get("org.opencontainers.image.title", "")
+            if title.endswith(".safetensors") and "digest" in layer:
+                safetensor_layers[title] = layer["digest"]
+
+        artifact = None
+        auth_header = resolve_auth_header(raw_token)
+        if auth_header:
+            try:
+                project, repo = split_repo_id(oci_repo)
+                artifact = harbor_get_artifact(auth_header, project, repo, ref.digest, endpoint=None)
+            except Exception:
+                pass  # timestamp metadata is best-effort; layer digests are the load-bearing part
+
+        committed_at = None
+        timestamp_source = None
+        if isinstance(artifact, dict) and artifact.get("push_time"):
+            committed_at, timestamp_source = artifact["push_time"], "harbor_artifact.push_time"
+        elif resp.headers.get("Last-Modified"):
+            committed_at, timestamp_source = resp.headers["Last-Modified"], "manifest_last_modified"
+
+        return {
+            "safetensor_layers": safetensor_layers,
+            "committed_at": committed_at,
+            "timestamp_source": timestamp_source,
+        }
+    except Exception:
+        # Fail OPEN: a metadata hiccup must never block a valid submission. The
+        # caller treats None as "cannot check", not "not a copy".
+        return None
+
+
 def upload_model_folder(
     folder_path: str | os.PathLike[str],
     repo: str,
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index 9344396..bfc65b5 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -122,6 +122,10 @@ def duel_ready(monkeypatch):
     monkeypatch.setattr(vmain, "SPEC", spec)
     monkeypatch.setattr(vmain, "CONSENSUS_DIGEST", spec.digest())
     monkeypatch.setattr(vmain, "PRESCREEN_ENABLED", False)
+    # The OCI copy check reaches the Hippius registry; off by default in these tests,
+    # which are about dispatch/settle policy. It has its own tests in test_copy_check.py,
+    # and one test in test_antiabuse.py turns it back on to prove it is wired.
+    monkeypatch.setattr(vmain, "COPY_CHECK_ENABLED", False)
     return spec
 
 
@@ -162,8 +166,10 @@ def __init__(self, monkeypatch, outcome, spec):
         self.dispatched: list[str] = []
         self.polled: list[str] = []
         self.jobs: dict[str, dict] = {}
+        self.cancelled: list[str] = []
         monkeypatch.setattr(vmain, "start_duel", self.start_duel)
         monkeypatch.setattr(vmain, "poll_duel", self.poll_duel)
+        monkeypatch.setattr(vmain, "cancel_duel", self.cancel_duel)
 
     async def start_duel(self, entry, king, block_hash):
         outcome = self._outcome(entry)
@@ -181,6 +187,9 @@ async def poll_duel(self, eval_id):
             raise outcome
         return outcome
 
+    async def cancel_duel(self, eval_id):
+        self.cancelled.append(eval_id)
+
     async def drive(self, state, store, entries, *, block, ticks=None):
         """Run enough ticks to settle + dispatch every challenger."""
         import leoma.app.validator.main as vmain
diff --git a/tests/unit/test_calibrate.py b/tests/unit/test_calibrate.py
new file mode 100644
index 0000000..b8c88e0
--- /dev/null
+++ b/tests/unit/test_calibrate.py
@@ -0,0 +1,152 @@
+"""The cross-GPU noise-measurement analysis.
+
+The number this produces is the one thing standing between Leoma's consensus
+converging and forking: is ``delta_threshold`` wider than the hardware noise, or not?
+Each test pins a property of the measurement, using synthetic per-box records so the
+pure analysis is exercised without a GPU.
+
+The experiment it analyzes: the SAME model, generated on N different GPU types, scored
+on the SAME clips with the SAME seed. Any per-clip distance difference between two
+boxes is pure hardware noise, because everything else is held identical. A self-duel
+between two boxes has a true mean advantage of zero, so whatever the bootstrap finds is
+the bias a marginal verdict could suffer.
+"""
+
+import numpy as np
+import pytest
+
+from leoma.eval.calibrate import analyze
+
+
+def _record(gpu, distances, *, model="u/leoma-k@sha256:aaa", corpus="v1", digests=None):
+    clips = []
+    for i, d in enumerate(distances):
+        clip = {"clip_id": f"clip-{i:04d}", "distance": float(d)}
+        if digests is not None:
+            clip["frames_digest"] = digests[i]
+        clips.append(clip)
+    return {"gpu": gpu, "model": model, "corpus_id": corpus, "clips": clips}
+
+
+class TestValidation:
+    def test_one_box_is_not_enough(self):
+        with pytest.raises(ValueError, match="at least 2"):
+            analyze([_record("H100", [0.1, 0.2])])
+
+    def test_different_models_are_rejected(self):
+        a = _record("H100", [0.1, 0.2], model="u/a@sha256:aaa")
+        b = _record("A100", [0.1, 0.2], model="u/b@sha256:bbb")
+        with pytest.raises(ValueError, match="different models"):
+            analyze([a, b])
+
+    def test_different_corpora_are_rejected(self):
+        a = _record("H100", [0.1, 0.2], corpus="v1")
+        b = _record("A100", [0.1, 0.2], corpus="v2")
+        with pytest.raises(ValueError, match="different corpora"):
+            analyze([a, b])
+
+
+class TestNoiseMeasurement:
+    def test_two_identical_boxes_have_zero_noise(self):
+        """If both boxes produced the same distances (perfect reproducibility), the
+        measured floor is exactly zero and any positive delta clears it."""
+        dists = [0.30, 0.42, 0.28, 0.35, 0.31]
+        r = analyze([_record("H100", dists), _record("A100", dists)],
+                    current_delta_threshold=0.0025)
+        assert r.max_abs_mu_hat == 0.0
+        assert r.max_abs_delta == 0.0
+        assert r.recommended_delta_threshold == 0.0
+        assert r.verdict.startswith("PASS")
+
+    def test_noise_shows_up_as_a_nonzero_floor(self):
+        rng = np.random.default_rng(0)
+        base = rng.uniform(0.2, 0.5, size=40)
+        noise = rng.normal(0, 0.01, size=40)   # 1e-2 per-clip hardware jitter
+        r = analyze([_record("H100", base), _record("A100", base + noise)],
+                    current_delta_threshold=0.0025)
+        assert r.max_abs_mu_hat > 0
+        assert r.max_abs_delta > 0
+        assert r.recommended_delta_threshold > 0
+
+    def test_the_recommendation_is_the_safety_multiple_of_the_floor(self):
+        rng = np.random.default_rng(1)
+        base = rng.uniform(0.2, 0.5, size=50)
+        r3 = analyze([_record("H100", base), _record("A100", base + rng.normal(0, 0.01, 50))],
+                     safety_factor=3.0)
+        # Doubling the safety factor doubles the recommendation (same underlying floor).
+        rng = np.random.default_rng(1)
+        base = rng.uniform(0.2, 0.5, size=50)
+        r6 = analyze([_record("H100", base), _record("A100", base + rng.normal(0, 0.01, 50))],
+                     safety_factor=6.0)
+        # Both are independently rounded to 6 decimals, so allow one rounding unit.
+        assert r6.recommended_delta_threshold == pytest.approx(2 * r3.recommended_delta_threshold, abs=2e-6)
+
+    def test_a_delta_below_the_floor_FAILS_loudly(self):
+        """The whole point of the tool: catch a delta_threshold that would let two
+        honest validators fork."""
+        rng = np.random.default_rng(2)
+        base = rng.uniform(0.2, 0.5, size=60)
+        big_noise = base + rng.normal(0, 0.05, size=60)   # noise >> a tiny delta
+        r = analyze([_record("H100", base), _record("A100", big_noise)],
+                    current_delta_threshold=0.0001)   # absurdly tight
+        assert r.verdict.startswith("FAIL")
+        assert "fork" in r.verdict
+
+    def test_a_delta_above_the_floor_PASSES(self):
+        rng = np.random.default_rng(3)
+        base = rng.uniform(0.2, 0.5, size=60)
+        tiny_noise = base + rng.normal(0, 1e-5, size=60)   # noise << delta
+        r = analyze([_record("H100", base), _record("A100", tiny_noise)],
+                    current_delta_threshold=0.0025)
+        assert r.verdict.startswith("PASS")
+
+
+class TestManyBoxes:
+    def test_all_pairs_are_compared(self):
+        rng = np.random.default_rng(4)
+        base = rng.uniform(0.2, 0.5, size=30)
+        boxes = [_record(f"gpu{i}", base + rng.normal(0, 0.01, 30)) for i in range(4)]
+        r = analyze(boxes)
+        assert r.n_boxes == 4
+        assert r.n_pairs == 6      # 4 choose 2
+
+    def test_the_floor_is_the_WORST_pair_not_the_average(self):
+        """One noisy box pair must drive the recommendation — consensus fails on the
+        worst case, not the mean."""
+        base = np.linspace(0.2, 0.5, 40)
+        quiet = base + 1e-6
+        loud = base + 0.03
+        r = analyze([_record("H100", base), _record("A100", quiet), _record("MI300", loud)])
+        # The H100-vs-MI300 (and A100-vs-MI300) pair dominates.
+        worst = max(p.abs_delta_max for p in r.pairs)
+        assert r.max_abs_delta == worst
+        assert worst > 0.02
+
+
+class TestBitReproducibility:
+    def test_identical_frame_digests_are_reported_as_fully_reproducible(self):
+        """If two boxes generated byte-identical frames, they are fully reproducible for
+        this model — the ideal, and worth surfacing distinctly from 'close distances'."""
+        dists = [0.3, 0.4, 0.35]
+        digs = ["sha256:d0", "sha256:d1", "sha256:d2"]
+        r = analyze([_record("H100", dists, digests=digs),
+                     _record("A100", dists, digests=digs)])
+        assert r.fully_reproducible_pairs == 1
+        assert r.pairs[0].bit_identical_clips == 3
+
+    def test_differing_frames_are_not_counted_reproducible(self):
+        r = analyze([_record("H100", [0.3, 0.4], digests=["sha256:a", "sha256:b"]),
+                     _record("A100", [0.3, 0.4], digests=["sha256:a", "sha256:X"])])
+        assert r.fully_reproducible_pairs == 0
+        assert r.pairs[0].bit_identical_clips == 1
+
+
+class TestDeterminism:
+    def test_the_analysis_is_reproducible(self):
+        rng = np.random.default_rng(5)
+        base = rng.uniform(0.2, 0.5, size=40)
+        boxes = [_record("H100", base), _record("A100", base + rng.normal(0, 0.01, 40))]
+        a = analyze(boxes, seed=7)
+        b = analyze(boxes, seed=7)
+        assert a.recommended_delta_threshold == b.recommended_delta_threshold
+        assert a.max_abs_mu_hat == b.max_abs_mu_hat
diff --git a/tests/unit/test_copy_check.py b/tests/unit/test_copy_check.py
new file mode 100644
index 0000000..ecf1527
--- /dev/null
+++ b/tests/unit/test_copy_check.py
@@ -0,0 +1,215 @@
+"""OCI-layer copy detection + earliest-author displacement.
+
+The free exact-digest check (`main._copies_a_king`) catches the crude copy — a
+different hotkey re-committing the king's *exact* digest. It misses the copy that
+matters: identical WEIGHTS repackaged with a changed README or tokenizer, which
+yields a new top-level digest but the SAME per-layer safetensor digests. On a
+content-addressed registry that is the king's weights in a disguise, and under the
+old gate it burned a full multi-hour duel before tying and losing.
+
+`check_model_copy` closes that for one manifest fetch (a few KB, no weight download),
+and while it is at it, displaces the king with a byte-identical model that was pushed
+to the registry EARLIER — the true original, front-run by whoever got crowned first.
+"""
+
+import leoma.app.validator.main as vmain
+from leoma.app.validator.copy_check import _parse_registry_timestamp, check_model_copy
+from leoma.app.validator.reveal_scan import ChallengerEntry
+from leoma.app.validator.state_store import JsonBucketStore, KingState
+
+from tests.unit.conftest import FakeEvalBox, FakeMinio
+
+KING_REPO = "u/leoma-king"
+KING_DIGEST = "sha256:" + "a" * 64
+LAYERS = {"transformer/model-00001.safetensors": "sha256:aaa",
+          "vae/model-00001.safetensors": "sha256:bbb"}
+
+
+def _info(layers, *, ts, src):
+    return {"safetensor_layers": dict(layers), "committed_at": ts, "timestamp_source": src}
+
+
+def _fetcher(mapping):
+    """Build a fetch(ref)->info stub keyed by digest."""
+    return lambda ref: mapping.get(ref.digest)
+
+
+class TestDecision:
+    def test_genuinely_different_weights_are_not_a_copy(self):
+        other = {"transformer/model-00001.safetensors": "sha256:XXX",
+                 "vae/model-00001.safetensors": "sha256:bbb"}
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts="2026-01-01T00:00:00Z", src="harbor_artifact.push_time"),
+            "sha256:" + "c" * 64: _info(other, ts="2026-02-01T00:00:00Z", src="harbor_artifact.push_time"),
+        })
+        assert check_model_copy("u/c", "sha256:" + "c" * 64, KING_REPO, KING_DIGEST, fetch=fetch) is None
+
+    def test_a_different_layer_COUNT_is_not_a_copy(self):
+        fewer = {"transformer/model-00001.safetensors": "sha256:aaa"}
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts="2026-01-01T00:00:00Z", src="harbor_artifact.push_time"),
+            "sha256:" + "c" * 64: _info(fewer, ts="2026-02-01T00:00:00Z", src="harbor_artifact.push_time"),
+        })
+        assert check_model_copy("u/c", "sha256:" + "c" * 64, KING_REPO, KING_DIGEST, fetch=fetch) is None
+
+    def test_identical_layers_committed_LATER_is_rejected(self):
+        """The repackaged copy the exact-digest check misses: same weights, new top
+        digest, pushed after the king."""
+        c = "sha256:" + "c" * 64
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts="2026-01-01T00:00:00Z", src="harbor_artifact.push_time"),
+            c: _info(LAYERS, ts="2026-02-01T00:00:00Z", src="harbor_artifact.push_time"),
+        })
+        v = check_model_copy("u/c", c, KING_REPO, KING_DIGEST, fetch=fetch)
+        assert v["action"] == "reject"
+        assert "identical OCI digests" in v["reason"]
+
+    def test_identical_layers_committed_EARLIER_displaces_the_king(self):
+        """The original author, front-run: byte-identical weights, earlier push."""
+        c = "sha256:" + "c" * 64
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts="2026-02-01T00:00:00Z", src="harbor_artifact.push_time"),
+            c: _info(LAYERS, ts="2026-01-01T00:00:00Z", src="harbor_artifact.push_time"),  # earlier
+        })
+        v = check_model_copy("u/c", c, KING_REPO, KING_DIGEST, fetch=fetch)
+        assert v["action"] == "crown_earlier"
+        assert "original author" in v["reason"]
+
+    def test_the_exact_reigning_king_is_rejected_without_a_fetch(self):
+        called = []
+        fetch = lambda ref: called.append(ref) or None
+        v = check_model_copy(KING_REPO, KING_DIGEST, KING_REPO, KING_DIGEST, fetch=fetch)
+        assert v["action"] == "reject"
+        assert called == [], "no metadata fetch needed for an exact re-commit"
+
+
+class TestFailSafeAndFailOpen:
+    def test_missing_timestamps_REJECT_never_crown(self):
+        """Fail-safe: identical weights but no registry timestamp => reject. Turning a
+        real author away is the safe direction; enthroning a plagiarist on backdated
+        metadata is not."""
+        c = "sha256:" + "c" * 64
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts=None, src=None),
+            c: _info(LAYERS, ts=None, src=None),
+        })
+        v = check_model_copy("u/c", c, KING_REPO, KING_DIGEST, fetch=fetch)
+        assert v["action"] == "reject"
+        assert "timestamps unavailable" in v["reason"]
+
+    def test_a_client_supplied_only_timestamp_does_not_earn_crown_earlier(self):
+        """If the earlier side has no registry source, it cannot displace — even though
+        its timestamp string is earlier. (The production fetcher never populates a
+        client annotation as committed_at; this guards the decision logic regardless.)"""
+        c = "sha256:" + "c" * 64
+        fetch = _fetcher({
+            KING_DIGEST: _info(LAYERS, ts="2026-02-01T00:00:00Z", src="harbor_artifact.push_time"),
+            c: _info(LAYERS, ts="2026-01-01T00:00:00Z", src=None),  # earlier but no source
+        })
+        v = check_model_copy("u/c", c, KING_REPO, KING_DIGEST, fetch=fetch)
+        assert v["action"] == "reject"
+
+    def test_a_metadata_hiccup_fails_OPEN(self):
+        """Fetch returns None (registry unreachable) => the check returns None => the
+        caller lets the model duel. A registry blip must never block a valid submission."""
+        fetch = lambda ref: None
+        assert check_model_copy("u/c", "sha256:" + "c" * 64, KING_REPO, KING_DIGEST, fetch=fetch) is None
+
+    def test_no_king_means_nothing_to_copy(self):
+        assert check_model_copy("u/c", "sha256:" + "c" * 64, "", "", fetch=lambda r: None) is None
+
+
+class TestTimestampParsing:
+    def test_iso_8601_with_z(self):
+        assert _parse_registry_timestamp("2026-01-01T00:00:00Z") is not None
+
+    def test_rfc_2822_last_modified(self):
+        # The manifest Last-Modified header form.
+        assert _parse_registry_timestamp("Wed, 01 Jan 2026 00:00:00 GMT") is not None
+
+    def test_earlier_really_compares_earlier_across_formats(self):
+        iso = _parse_registry_timestamp("2026-01-01T00:00:00Z")
+        rfc = _parse_registry_timestamp("Wed, 01 Feb 2026 00:00:00 GMT")
+        assert iso < rfc
+
+    def test_garbage_is_none(self):
+        assert _parse_registry_timestamp("not a date") is None
+        assert _parse_registry_timestamp(None) is None
+
+
+class TestWiredIntoDispatch:
+    async def test_a_repackaged_copy_never_reaches_the_GPU(self, monkeypatch, duel_ready):
+        """The whole point: a copy that changed only its README costs one manifest
+        fetch, not a multi-hour duel."""
+        monkeypatch.setattr(vmain, "COPY_CHECK_ENABLED", True)
+        c = "sha256:" + "c" * 64
+        monkeypatch.setattr(vmain, "check_model_copy",
+                            lambda cr, cd, kr, kd: {"action": "reject", "reason": "repackaged king"})
+
+        box = FakeEvalBox(monkeypatch, lambda e: AssertionError("must not dispatch"), duel_ready)
+        st = KingState()
+        st.king = {"hotkey": "5KING", "model_repo": KING_REPO, "model_digest": KING_DIGEST, "reign_number": 1}
+        store = JsonBucketStore(FakeMinio(), "own", backoff=0)
+        entry = ChallengerEntry(hotkey="5thief", model_repo="u/leoma-copy", model_digest=c, block=100)
+
+        await box.drive(st, store, [entry], block=200, ticks=1)
+
+        assert box.dispatched == []
+        key = vmain._seen_key(entry.hotkey, entry.model_digest)
+        assert st.attempts[key]["last_reason"] == "copy_of_king"
+        assert st.duels["5thief"]["strikes"] == 1
+
+    async def test_an_earlier_author_takes_the_crown_with_no_duel(self, monkeypatch, duel_ready):
+        monkeypatch.setattr(vmain, "COPY_CHECK_ENABLED", True)
+        c = "sha256:" + "c" * 64
+        monkeypatch.setattr(vmain, "check_model_copy", lambda cr, cd, kr, kd: {
+            "action": "crown_earlier", "reason": "earlier push",
+            "challenger_committed_at": "2026-01-01T00:00:00Z", "king_committed_at": "2026-02-01T00:00:00Z",
+        })
+
+        box = FakeEvalBox(monkeypatch, lambda e: AssertionError("must not duel a proven copy"), duel_ready)
+        st = KingState()
+        st.king = {"hotkey": "5KING", "model_repo": KING_REPO, "model_digest": KING_DIGEST, "reign_number": 1}
+        store = JsonBucketStore(FakeMinio(), "own", backoff=0)
+        entry = ChallengerEntry(hotkey="5author", model_repo="u/leoma-orig", model_digest=c, block=100)
+
+        await box.drive(st, store, [entry], block=200, ticks=1)
+
+        assert box.dispatched == []                       # no duel
+        assert st.king["hotkey"] == "5author"             # the original author reigns
+        assert st.king["model_digest"] == c
+        assert st.king["reign_number"] == 2               # a genuine dethrone
+        assert st.history[0]["verdict"] == "crown_earlier"
+        assert st.stats["accepted"] == 1
+
+    async def test_a_metadata_hiccup_lets_the_model_duel(self, monkeypatch, duel_ready):
+        """Fail-open at the call site too: None => proceed to the normal duel."""
+        monkeypatch.setattr(vmain, "COPY_CHECK_ENABLED", True)
+        monkeypatch.setattr(vmain, "check_model_copy", lambda *a: None)
+
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st = KingState()
+        st.king = {"hotkey": "5KING", "model_repo": KING_REPO, "model_digest": KING_DIGEST, "reign_number": 1}
+        store = JsonBucketStore(FakeMinio(), "own", backoff=0)
+        entry = ChallengerEntry(hotkey="5new", model_repo="u/leoma-new",
+                                model_digest="sha256:" + "n" * 64, block=100)
+
+        await box.drive(st, store, [entry], block=200, ticks=1)
+        assert box.dispatched == ["5new"]
+
+    async def test_the_check_crashing_does_not_crash_the_tick(self, monkeypatch, duel_ready):
+        monkeypatch.setattr(vmain, "COPY_CHECK_ENABLED", True)
+
+        def boom(*a):
+            raise RuntimeError("registry exploded")
+
+        monkeypatch.setattr(vmain, "check_model_copy", boom)
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st = KingState()
+        st.king = {"hotkey": "5KING", "model_repo": KING_REPO, "model_digest": KING_DIGEST, "reign_number": 1}
+        store = JsonBucketStore(FakeMinio(), "own", backoff=0)
+        entry = ChallengerEntry(hotkey="5new", model_repo="u/leoma-new",
+                                model_digest="sha256:" + "n" * 64, block=100)
+
+        await box.drive(st, store, [entry], block=200, ticks=1)   # must not raise
+        assert box.dispatched == ["5new"]                          # failed open
diff --git a/tests/unit/test_cuda_selfkill.py b/tests/unit/test_cuda_selfkill.py
new file mode 100644
index 0000000..6391f9a
--- /dev/null
+++ b/tests/unit/test_cuda_selfkill.py
@@ -0,0 +1,154 @@
+"""CUDA-context-poison self-kill.
+
+Once a CUDA context is corrupted — an illegal memory access, a device-side assert, a
+cuBLAS execution failure — the allocator and every stream on the process are poisoned,
+and every subsequent load/generate keeps raising against the dead context. The
+watchdog's `os._exit` only fires when a duel *stalls*; a duel that FAILS FAST on a CUDA
+fault would release the lock and hand the next challenger a poisoned box, mis-rejecting
+an honest model as broken. For Leoma that wastes *hours* of the next duel, not minutes.
+
+The only recovery is to exit and let the supervisor restart with a fresh context. This
+is safe precisely because the chain is the queue: the restart loses at most the current
+duel, which the validator re-dispatches. Ported from Teutonic's self-kill.
+"""
+
+import threading
+import time
+
+import pytest
+from starlette.testclient import TestClient
+
+import leoma.eval_server as es
+from leoma.eval_server import create_app, is_cuda_fatal
+
+from .conftest import pinned_spec
+
+
+SPEC = pinned_spec()
+REQ = dict(
+    king_repo="u/leoma-k", king_digest="sha256:" + "a" * 64,
+    challenger_repo="u/leoma-c", challenger_digest="sha256:" + "b" * 64,
+    block_hash="0xabc", hotkey="5C7L",
+    spec=SPEC.model_dump(mode="json"), consensus_digest=SPEC.digest(),
+)
+
+
+@pytest.fixture
+def caught_exit(monkeypatch):
+    """Replace os._exit with a recorder and reset the idempotency latch."""
+    monkeypatch.setattr(es, "_self_kill_scheduled", threading.Event())
+    monkeypatch.setattr(es, "CUDA_FATAL_EXIT_DELAY_S", 0.0)
+    calls: list[int] = []
+    done = threading.Event()
+    monkeypatch.setattr(es.os, "_exit", lambda code: (calls.append(code), done.set()))
+    return calls, done
+
+
+class TestIsCudaFatal:
+    @pytest.mark.parametrize("msg", [
+        "RuntimeError: CUDA error: an illegal memory access was encountered",
+        "cudaErrorIllegalAddress",
+        "CUDA error: device-side assert triggered",
+        "CUDA error: misaligned address",
+        "CUBLAS_STATUS_EXECUTION_FAILED when calling cublasGemmEx",
+        "cuDNN error: CUDNN_STATUS_EXECUTION_FAILED",
+        "Bus error",
+        "Segmentation fault",
+    ])
+    def test_fatal_tokens_are_recognized(self, msg):
+        assert is_cuda_fatal(msg) is True
+
+    def test_case_insensitive(self):
+        assert is_cuda_fatal("cuda error: MISALIGNED ADDRESS") is True
+
+    @pytest.mark.parametrize("msg", [
+        "CUDA out of memory. Tried to allocate 2.00 GiB",   # recoverable — empty_cache + retry
+        "RepositoryNotFoundError: repo missing",
+        "connection reset by peer",
+        "ValueError: generation too short",
+        "",
+    ])
+    def test_recoverable_and_unrelated_errors_are_NOT_fatal(self, msg):
+        # OOM especially must not self-kill: it is transient and the box recovers.
+        assert is_cuda_fatal(msg) is False
+
+    def test_accepts_an_exception_object(self):
+        assert is_cuda_fatal(RuntimeError("CUDA error: an illegal memory access")) is True
+
+
+class TestScheduleSelfKill:
+    def test_it_eventually_exits_with_the_configured_code(self, caught_exit):
+        calls, done = caught_exit
+        es.schedule_self_kill("test")
+        assert done.wait(timeout=5)
+        assert calls == [es.CUDA_FATAL_EXIT_CODE]
+
+    def test_it_is_idempotent(self, caught_exit):
+        calls, done = caught_exit
+        es.schedule_self_kill("first")
+        es.schedule_self_kill("second")
+        es.schedule_self_kill("third")
+        assert done.wait(timeout=5)
+        time.sleep(0.05)
+        assert calls == [es.CUDA_FATAL_EXIT_CODE], "scheduled the exit more than once"
+
+
+class TestWiredIntoTheWorker:
+    def _events(self, client, eval_id):
+        out = []
+        with client.stream("GET", f"/eval/{eval_id}/stream") as s:
+            for line in s.iter_lines():
+                if line and line.startswith("data: "):
+                    import json
+                    out.append(json.loads(line[6:]))
+        return out
+
+    def test_a_cuda_fatal_duel_reports_its_error_AND_self_kills(self, caught_exit):
+        """Two things must both happen: the validator still gets a terminal error event
+        (so it can retry), and the box schedules its own restart."""
+        calls, done = caught_exit
+
+        def poisoned(req, emit, cancel):
+            emit({"phase": "load"})
+            raise RuntimeError("CUDA error: an illegal memory access was encountered")
+
+        c = TestClient(create_app(runner=poisoned))
+        eval_id = c.post("/eval", json=REQ).json()["eval_id"]
+        events = self._events(c, eval_id)
+
+        # The validator is told — as a normal terminal error, so it retries elsewhere.
+        assert events[-1]["phase"] == "error"
+        assert "illegal memory access" in events[-1]["error"]
+        # And the box is taking itself down for a fresh context.
+        assert done.wait(timeout=5)
+        assert calls == [es.CUDA_FATAL_EXIT_CODE]
+
+    def test_an_ordinary_duel_failure_does_NOT_self_kill(self, caught_exit):
+        """A 404 or a bad model is not a poisoned context — the box must keep serving."""
+        calls, done = caught_exit
+
+        def broken_model(req, emit, cancel):
+            raise RuntimeError("RepositoryNotFoundError: repo missing")
+
+        c = TestClient(create_app(runner=broken_model))
+        eval_id = c.post("/eval", json=REQ).json()["eval_id"]
+        self._events(c, eval_id)
+
+        assert not done.wait(timeout=0.5)
+        assert calls == []
+        # And the box is still usable.
+        assert c.get("/health").json()["busy"] is False
+
+    def test_an_oom_does_NOT_self_kill(self, caught_exit):
+        """The one that would be easy to get wrong: OOM is transient, not fatal."""
+        calls, done = caught_exit
+
+        def oom(req, emit, cancel):
+            raise RuntimeError("CUDA out of memory. Tried to allocate 2.00 GiB")
+
+        c = TestClient(create_app(runner=oom))
+        eval_id = c.post("/eval", json=REQ).json()["eval_id"]
+        self._events(c, eval_id)
+
+        assert not done.wait(timeout=0.5)
+        assert calls == []
diff --git a/tests/unit/test_inflight_guillotine.py b/tests/unit/test_inflight_guillotine.py
new file mode 100644
index 0000000..630fbe6
--- /dev/null
+++ b/tests/unit/test_inflight_guillotine.py
@@ -0,0 +1,140 @@
+"""The validator-side stuck-duel backstop.
+
+The eval box has its own forward-progress watchdog, and normally it fires first. But
+Leoma moved the wall-clock bound entirely onto the box, so if the box 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 — one stuck duel would hold the
+single in-flight slot forever and no other challenger would ever be dispatched.
+
+This is the guillotine: abandon a duel that has been in flight far past any plausible
+wall clock, and free the slot. It is deliberately generous (a real duel of two 14B
+models runs hours), so it only fires on a genuinely hung box.
+"""
+
+import leoma.app.validator.main as vmain
+from leoma.app.validator.reveal_scan import ChallengerEntry
+from leoma.app.validator.state_store import JsonBucketStore, KingState
+
+from tests.unit.conftest import FakeEvalBox, FakeMinio
+
+KING_DIGEST = "sha256:" + "k" * 64
+
+
+def _store():
+    return JsonBucketStore(FakeMinio(), "own", backoff=0)
+
+
+def _state():
+    st = KingState()
+    st.king = {"hotkey": "5KING", "model_repo": "u/leoma-king",
+               "model_digest": KING_DIGEST, "reign_number": 1}
+    return st
+
+
+def _entry(hotkey="5a", digest="sha256:" + "a" * 64, block=100):
+    return ChallengerEntry(hotkey=hotkey, model_repo="u/leoma-a", model_digest=digest, block=block)
+
+
+class TestGuillotine:
+    async def test_a_running_duel_within_the_bound_is_left_alone(self, monkeypatch, duel_ready):
+        monkeypatch.setattr(vmain, "MAX_INFLIGHT_BLOCKS", 1000)
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st, store = _state(), _store()
+        e = _entry()
+
+        await box.drive(st, store, [e], block=200, ticks=1)   # dispatched at 200
+        # Poll 500 blocks later — well inside the 1000-block bound.
+        await vmain.settle_inflight(_Sub(), None, st, {}, store, block=700)
+
+        assert st.inflight is not None, "a duel within the bound was abandoned"
+        assert box.cancelled == []
+
+    async def test_a_duel_past_the_bound_is_abandoned_and_the_slot_freed(self, monkeypatch, duel_ready):
+        monkeypatch.setattr(vmain, "MAX_INFLIGHT_BLOCKS", 1000)
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st, store = _state(), _store()
+        e = _entry()
+
+        await box.drive(st, store, [e], block=200, ticks=1)   # dispatched at 200
+        eval_id = st.inflight["eval_id"]
+
+        # Poll 1500 blocks later — past the 1000-block bound.
+        free = await vmain.settle_inflight(_Sub(), None, st, {}, store, block=1700)
+
+        assert free is True                       # the slot is free for the next dispatch
+        assert st.inflight is None
+        assert eval_id in box.cancelled           # we asked the box to stop burning GPU
+        key = vmain._seen_key(e.hotkey, e.model_digest)
+        assert st.attempts[key]["last_reason"] == "inflight_timeout"
+
+    async def test_the_stuck_box_is_never_blamed_on_the_miner(self, monkeypatch, duel_ready):
+        """A hung box is infrastructure, not the challenger. TRANSIENT + backoff, never
+        a strike — so a one-off wedge retries, but a model that hangs EVERY time still
+        exhausts its attempt budget and quarantines."""
+        monkeypatch.setattr(vmain, "MAX_INFLIGHT_BLOCKS", 100)
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st, store = _state(), _store()
+        e = _entry()
+
+        await box.drive(st, store, [e], block=200, ticks=1)
+        key = vmain._seen_key(e.hotkey, e.model_digest)
+
+        await vmain.settle_inflight(_Sub(), None, st, {}, store, block=1000)
+        assert st.attempts[key]["last_class"] == "transient"
+        assert st.duels.get(e.hotkey, {}).get("strikes", 0) == 0
+
+    async def test_after_abandon_the_next_challenger_is_dispatched(self, monkeypatch, duel_ready):
+        """The whole point: freeing the slot lets someone else run."""
+        monkeypatch.setattr(vmain, "MAX_INFLIGHT_BLOCKS", 100)
+        outcomes = {"5stuck": {"status": "running"}, "5next": {"status": "running"}}
+        box = FakeEvalBox(monkeypatch, lambda e: outcomes[e.hotkey], duel_ready)
+        st, store = _state(), _store()
+        stuck = _entry("5stuck", "sha256:" + "a" * 64)
+        nxt = _entry("5next", "sha256:" + "b" * 64)
+
+        # Tick 1 at block 200: dispatch 5stuck.
+        await vmain.process_challengers(_Sub(), None, st, {}, store, [stuck, nxt], 200)
+        assert st.inflight["hotkey"] == "5stuck"
+
+        # Tick 2 far later: 5stuck is guillotined, then 5next dispatched in the same tick.
+        await vmain.process_challengers(_Sub(), None, st, {}, store, [stuck, nxt], 2000)
+        assert st.inflight is not None and st.inflight["hotkey"] == "5next"
+
+    async def test_a_slot_with_no_dispatched_block_does_not_crash(self, monkeypatch, duel_ready):
+        """A slot persisted by an older build has no dispatched_block; treat it as
+        just-dispatched (age 0) rather than tripping the guillotine or raising."""
+        monkeypatch.setattr(vmain, "MAX_INFLIGHT_BLOCKS", 100)
+        box = FakeEvalBox(monkeypatch, lambda e: {"status": "running"}, duel_ready)
+        st, store = _state(), _store()
+        st.inflight = {
+            "eval_id": "eval-old", "hotkey": "5a", "model_repo": "u/leoma-a",
+            "model_digest": "sha256:" + "a" * 64, "block": 100, "king_digest": KING_DIGEST,
+            # no dispatched_block
+        }
+        box.jobs["eval-old"] = {"status": "running"}
+
+        free = await vmain.settle_inflight(_Sub(), None, st, {}, store, block=999999)
+        assert free is False              # age defaults to 0 -> not stuck
+        assert st.inflight is not None
+
+
+class _Sub:
+    async def get_block_hash(self, block):
+        return f"0x{block:064x}"
+
+    async def get_current_block(self):
+        return 1000
+
+    async def blocks_since_last_update(self, netuid, uid):
+        return 10_000
+
+    async def weights_rate_limit(self, netuid):
+        return 100
+
+    async def set_weights(self, **kwargs):
+        return True, "ok"
+
+    async def metagraph(self, netuid):
+        class M:
+            hotkeys: list = []
+        return M()