diff --git a/docs/TESTNET_RUNBOOK.md b/docs/TESTNET_RUNBOOK.md
new file mode 100644
index 0000000..be87129
--- /dev/null
+++ b/docs/TESTNET_RUNBOOK.md
@@ -0,0 +1,145 @@
+# Testnet dress-rehearsal runbook
+
+A repeatable rehearsal that proves the subnet does the right thing **before** mainnet.
+Every step ends in an assertion, not an eyeball — `leoma preflight` gates the launch,
+and `leoma smoke` confirms each scenario was actually exercised and handled correctly.
+
+Run this on testnet with at least one validator + one eval box (a GPU) and a handful of
+miner hotkeys you control.
+
+---
+
+## 0. Prerequisites — pin the consensus surface
+
+The subnet deliberately burns 100% to UID 0 until these are pinned. This is not optional.
+
+1. **Pick the base-model revision** and pin `chain.toml [seed].seed_digest` (the genesis
+   king) to the exact Wan2.1-I2V-14B-480P revision you will run — either a Hippius OCI
+   digest (`sha256:<64hex>`) or a HuggingFace commit SHA (`hf:<40hex>`). `preflight`
+   rejects anything else as unresolvable.
+2. **Build and publish the corpus:**
+   ```bash
+   leoma corpus build-manifest --corpus-id leoma-testnet-v1   # decides windows, hashes truth
+   leoma corpus publish-manifest manifest.json                # prints the digest
+   ```
+   Paste the printed digest into `chain.toml [corpus].manifest_digest`.
+3. **Verify each eval box** decodes the corpus byte-identically to the manifest:
+   ```bash
+   leoma corpus verify --sample 4
+   ```
+   A box that fails this must not duel — its distances would not be reproducible.
+
+## 1. Calibrate `delta_threshold` (the load-bearing measurement)
+
+This is the single largest open consensus risk. On **each GPU type** in the fleet:
+
+```bash
+leoma calibrate generate --gpu <label> -o box-<label>.json
+```
+
+Then, once, compare them all:
+
+```bash
+leoma calibrate analyze box-*.json
+```
+
+- **PASS** → the current `delta_threshold` clears the measured cross-GPU noise floor.
+  Proceed.
+- **FAIL** → `delta_threshold` is *below* the noise floor. Two honest validators can
+  fork. Raise it to the recommended value **or**, if the recommendation is implausibly
+  large, treat it as a signal that LPIPS-on-generated-frames is too noisy for
+  cross-hardware consensus and make a structural decision (a more reproducible metric,
+  or pin the fleet to one GPU class). **Do not launch through a FAIL.**
+
+## 2. Preflight — the launch gate
+
+On the validator box, with `EVAL_SERVER_URL`, `R2_OWN_BUCKET`, `WALLET_NAME`,
+`HOTKEY_NAME` set:
+
+```bash
+leoma preflight
+```
+
+It exits non-zero (and says exactly why) if the seed or corpus is unpinned, the
+consensus surface is invalid, or the eval box is on a different `chain.toml` / scoring
+code than the validator. **Gate your launch script on it:**
+
+```bash
+leoma preflight && leoma serve
+```
+
+Running several eval-server processes (one per GPU pair on an 8×H100 box)? Set
+`EVAL_SERVER_URLS` (comma-separated) instead of the single `EVAL_SERVER_URL` —
+`preflight` checks every configured URL independently and labels each finding by its
+box, so one stale server can't hide behind a healthy sibling.
+
+## 3. Start the services
+
+```bash
+# GPU box(es) — one process per pair of GPUs to duel on, each pinned via
+# LEOMA_KING_DEVICE/LEOMA_CHALLENGER_DEVICE if running more than one on the same host
+leoma servers eval-server           # binds 127.0.0.1; validator reaches it over an SSH tunnel
+
+# validator box — set EVAL_SERVER_URLS to the comma-separated list if running more than one
+leoma serve
+```
+
+## 4. Drive the scenarios
+
+Submit each of these as a miner (`leoma miner push` + `leoma miner commit`) and let the
+validator pick it up. The goal is to exercise every handling path once.
+
+| # | Scenario | How to produce it | Expected outcome |
+|---|----------|-------------------|------------------|
+| A | **Genuine crown** | A model that actually beats the genesis king | Crowned; king chain grows; weights shift |
+| B | **Fair rejection** | A model weaker than the king | Scored, `lcb < delta`, king holds |
+| C | **Broken repo (quarantine)** | Commit a reveal pointing at a non-existent repo | `error` row, `model_not_found`; quarantined after 2 sightings; **later challengers still run** |
+| D | **Wrong architecture** | A model whose `transformer/config.json` shape differs from the base | Rejected pre-dispatch (`arch_mismatch`) in ~seconds, no GPU spent |
+| E | **Copy of the king** | Re-upload the king's weights under a new hotkey (change only the README) | Rejected pre-duel (`copy_of_king`); no multi-hour duel |
+| F | **Freeze cheat** | A model that emits the conditioning frame repeated | Rejected by the freeze gate (`FROZE OUT`) even if it beat a weak king |
+
+## 5. Smoke — assert the outcomes
+
+Point `smoke` at the validator's published `dashboard.json`:
+
+```bash
+leoma smoke https://<your-state-bucket>/dashboard.json
+```
+
+It reports which rehearsal scenarios have been observed and exits non-zero until every
+one has. Re-run it as you drive more scenarios; a clean run means:
+
+- ✓ a challenger beat the king and was crowned
+- ✓ a challenger was scored and lost fairly
+- ✓ a broken model was recorded as an error (not silently dropped)
+- ✓ a copy of the king was rejected pre-duel
+- ✓ a freeze cheat was rejected by the gate
+- ✓ the validator is dueling, not degraded
+
+Check the **dashboard** during a real duel too — the "in the arena" panel should show
+the live challenger-vs-king, and the quality-over-reigns chart should show models
+sitting well below the dashed freeze cheat floor.
+
+## 6. Resilience spot-checks (optional but recommended)
+
+- **Restart the validator mid-duel.** It should re-attach to the in-flight slot and
+  settle the same duel, not orphan it.
+- **Kill the eval box mid-duel.** The validator should classify it transient and retry;
+  the box should come back with a fresh CUDA context.
+- **Point one eval box at a stale `chain.toml`.** `preflight` and the validator's
+  dispatch preflight should both refuse it (`consensus_mismatch`). With a single
+  configured eval server the dashboard should show the `degraded` reason; with several
+  configured (`EVAL_SERVER_URLS`), the stale one should simply be skipped in favor of a
+  healthy sibling — the validator should **not** show `degraded` as long as at least one
+  configured box is healthy.
+
+---
+
+## Definition of done
+
+- `leoma calibrate analyze` returns **PASS** for the fleet's `delta_threshold`.
+- `leoma preflight` exits 0 on every validator.
+- `leoma smoke` reports **all** scenarios observed.
+- The dashboard shows a live duel and the cheat-floor chart.
+
+Only then is the subnet ready for mainnet.
diff --git a/leoma/app/preflight.py b/leoma/app/preflight.py
new file mode 100644
index 0000000..809d2dc
--- /dev/null
+++ b/leoma/app/preflight.py
@@ -0,0 +1,242 @@
+"""Is this validator actually ready to launch? A hard gate, not a hope.
+
+The subnet has several pins that must be set before it does anything useful, and each
+one fails *safe but silent*: an unpinned seed digest, an unpinned corpus, or an eval
+box on a stale config all make the validator burn to UID 0 rather than crown. That is
+the correct behavior, but an operator who flips the switch and walks away would see a
+dead subnet with no single place that says *why*.
+
+``run_preflight`` is that single place. It runs every readiness check, classifies each
+as pass / warn / fail, and returns an overall verdict. ``fail`` means "the validator
+will not crown anyone until you fix this"; ``warn`` means "this will work but you
+probably didn't mean it" (e.g. no eval server configured to check against).
+
+The checks are **pure functions of already-fetched inputs** — the CLI does the I/O
+(HTTP to the eval box, a HEAD on the corpus) and hands the results here — so the
+decision logic is unit-testable with no network, GPU, or chain.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import NamedTuple, Optional
+
+from leoma.infra.model_store import DIGEST_RE
+
+PASS = "pass"
+WARN = "warn"
+FAIL = "fail"
+
+
+@dataclass(frozen=True)
+class CheckResult:
+    name: str
+    status: str
+    detail: str
+
+    @property
+    def ok(self) -> bool:
+        return self.status != FAIL
+
+
+@dataclass(frozen=True)
+class PreflightReport:
+    checks: tuple[CheckResult, ...]
+
+    @property
+    def ready(self) -> bool:
+        """True when nothing is FAIL. Warnings do not block launch."""
+        return all(c.ok for c in self.checks)
+
+    @property
+    def failures(self) -> tuple[CheckResult, ...]:
+        return tuple(c for c in self.checks if c.status == FAIL)
+
+    @property
+    def warnings(self) -> tuple[CheckResult, ...]:
+        return tuple(c for c in self.checks if c.status == WARN)
+
+
+def check_seed(seed_digest: str) -> CheckResult:
+    digest = (seed_digest or "").strip()
+    if not digest:
+        return CheckResult(
+            "seed_digest", FAIL,
+            "chain.toml [seed].seed_digest is empty — with no genesis king the subnet burns "
+            "100% to UID 0. Pin the exact base-model revision before launch.",
+        )
+    if not DIGEST_RE.match(digest):
+        return CheckResult(
+            "seed_digest", FAIL,
+            f"chain.toml [seed].seed_digest ({digest[:23]}…) is not a recognized digest — "
+            "expected 'sha256:<64hex>' (Hippius) or 'hf:<40hex>' (HuggingFace commit SHA). "
+            "A malformed pin means the genesis king can never resolve.",
+        )
+    return CheckResult("seed_digest", PASS, f"genesis king pinned ({digest[:23]}…)")
+
+
+def check_corpus_pin(corpus_pinned: bool, manifest_digest: str) -> CheckResult:
+    if corpus_pinned:
+        return CheckResult("corpus_pin", PASS, f"corpus manifest pinned ({manifest_digest[:23]}…)")
+    return CheckResult(
+        "corpus_pin", FAIL,
+        "chain.toml [corpus].manifest_digest is empty — the duel exam is not reproducible "
+        "and the validator refuses to duel. Publish a manifest (`leoma corpus publish-manifest`) "
+        "and pin its digest.",
+    )
+
+
+def check_consensus_digest(consensus_digest: str) -> CheckResult:
+    # If chain_config imported at all, SPEC validated and this exists — so this is really
+    # a "surface the digest so the operator can compare it across boxes" check.
+    if consensus_digest and consensus_digest.startswith("sha256:"):
+        return CheckResult("consensus_digest", PASS, consensus_digest)
+    return CheckResult("consensus_digest", FAIL, "consensus surface did not produce a digest")
+
+
+def check_corpus_reachable(fetched_digest: Optional[str], pinned_digest: str, error: Optional[str]) -> CheckResult:
+    """Given the digest of the manifest actually fetched from the bucket, does it match?"""
+    if error:
+        return CheckResult("corpus_fetch", WARN, f"could not fetch the corpus manifest to verify it: {error}")
+    if not fetched_digest:
+        return CheckResult("corpus_fetch", WARN, "corpus manifest not checked (no bucket credentials)")
+    if fetched_digest == pinned_digest:
+        return CheckResult("corpus_fetch", PASS, "published manifest matches the pinned digest")
+    return CheckResult(
+        "corpus_fetch", FAIL,
+        f"the bucket's manifest ({fetched_digest[:19]}…) does NOT match the pinned digest "
+        f"({pinned_digest[:19]}…). Republish, or fix the pin.",
+    )
+
+
+def check_eval_server(
+    health: Optional[dict],
+    our_consensus_digest: str,
+    our_eval_code_digest: str,
+    error: Optional[str] = None,
+    *,
+    name: str = "eval_server",
+) -> CheckResult:
+    """Given one eval box's /health, do its consensus + code digests match ours?
+
+    ``name`` lets the caller disambiguate several servers (e.g. ``eval_server[url]``)
+    when checking a whole ``EVAL_SERVER_URLS`` fleet instead of a single box.
+    """
+    if error:
+        return CheckResult(name, WARN, f"eval server not reachable ({error}); skipped")
+    if health is None:
+        return CheckResult(name, WARN, "no eval server configured to check against (set EVAL_SERVER_URL(S))")
+
+    theirs_consensus = health.get("consensus_digest")
+    theirs_code = health.get("eval_code_digest")
+    if theirs_consensus != our_consensus_digest:
+        return CheckResult(
+            name, FAIL,
+            f"eval box pins a DIFFERENT consensus surface (box {str(theirs_consensus)[:19]}…, "
+            f"validator {our_consensus_digest[:19]}…) — one of you is on a stale chain.toml.",
+        )
+    if theirs_code is None:
+        # A current box always reports this field (see eval_server.py's /health). A box
+        # missing it entirely is running a build old enough to predate the field — we
+        # have literally no evidence its scoring code matches, so this must not read as
+        # a silent PASS.
+        return CheckResult(
+            name, WARN,
+            "eval box's /health did not report eval_code_digest (stale build?) — its "
+            "scoring code could not be verified against this validator's.",
+        )
+    if theirs_code != our_eval_code_digest:
+        return CheckResult(
+            name, FAIL,
+            f"eval box runs DIFFERENT scoring code (box {str(theirs_code)[:19]}…, "
+            f"validator {our_eval_code_digest[:19]}…) — its distances would not be reproducible.",
+        )
+    return CheckResult(name, PASS, "eval box matches this validator's consensus surface + code")
+
+
+class EvalServerProbe(NamedTuple):
+    """One configured server's raw /health result, ready for ``check_eval_servers``."""
+    url: str
+    health: Optional[dict]
+    error: Optional[str] = None
+
+
+def check_eval_servers(
+    probes: tuple[EvalServerProbe, ...],
+    our_consensus_digest: str,
+    our_eval_code_digest: str,
+) -> tuple[CheckResult, ...]:
+    """One :func:`check_eval_server` result per configured ``EVAL_SERVER_URLS`` entry.
+
+    A single-server validator gets exactly one ``eval_server`` check, unchanged. A
+    multi-server validator gets one check per URL — silently checking only the first
+    configured server (or none at all) would leave the rest of the fleet unverified.
+    """
+    if not probes:
+        return (check_eval_server(None, our_consensus_digest, our_eval_code_digest),)
+    if len(probes) == 1:
+        p = probes[0]
+        return (check_eval_server(p.health, our_consensus_digest, our_eval_code_digest, p.error),)
+    return tuple(
+        check_eval_server(
+            p.health, our_consensus_digest, our_eval_code_digest, p.error,
+            name=f"eval_server[{p.url}]",
+        )
+        for p in probes
+    )
+
+
+def check_state_bucket(own_bucket: Optional[str]) -> CheckResult:
+    if own_bucket and own_bucket.strip():
+        return CheckResult("state_bucket", PASS, f"king state persists to {own_bucket}")
+    return CheckResult(
+        "state_bucket", FAIL,
+        "R2_OWN_BUCKET is not set — the validator cannot persist king state and will refuse to run.",
+    )
+
+
+def check_wallet(wallet_name: Optional[str], hotkey_name: Optional[str]) -> CheckResult:
+    if wallet_name and hotkey_name:
+        return CheckResult("wallet", PASS, f"{wallet_name}/{hotkey_name}")
+    return CheckResult("wallet", WARN, "wallet/hotkey not both set (using defaults)")
+
+
+def run_preflight(
+    *,
+    seed_digest: str,
+    corpus_pinned: bool,
+    manifest_digest: str,
+    consensus_digest: str,
+    eval_code_digest: str,
+    own_bucket: Optional[str],
+    wallet_name: Optional[str],
+    hotkey_name: Optional[str],
+    corpus_fetched_digest: Optional[str] = None,
+    corpus_error: Optional[str] = None,
+    eval_servers: tuple[EvalServerProbe, ...] = (),
+) -> PreflightReport:
+    """Assemble every readiness check into one verdict. Pure — the caller does the I/O.
+
+    ``eval_servers`` is one probe per configured ``EVAL_SERVER_URLS`` entry (empty when
+    none are configured) — a single-server validator still gets exactly one
+    ``eval_server`` check; a multi-server one gets one per URL, so a stale box can't
+    hide behind a healthy sibling.
+    """
+    checks = [
+        check_seed(seed_digest),
+        check_corpus_pin(corpus_pinned, manifest_digest),
+        check_consensus_digest(consensus_digest),
+        check_corpus_reachable(corpus_fetched_digest, manifest_digest, corpus_error),
+        *check_eval_servers(eval_servers, consensus_digest, eval_code_digest),
+        check_state_bucket(own_bucket),
+        check_wallet(wallet_name, hotkey_name),
+    ]
+    return PreflightReport(tuple(checks))
+
+
+__all__ = [
+    "PASS", "WARN", "FAIL",
+    "CheckResult", "PreflightReport", "EvalServerProbe", "run_preflight",
+    "check_seed", "check_corpus_pin", "check_consensus_digest",
+    "check_corpus_reachable", "check_eval_server", "check_eval_servers",
+    "check_state_bucket", "check_wallet",
+]
diff --git a/leoma/app/smoke.py b/leoma/app/smoke.py
new file mode 100644
index 0000000..ab56e76
--- /dev/null
+++ b/leoma/app/smoke.py
@@ -0,0 +1,152 @@
+"""Did the subnet actually do the right thing? Scenario matchers over dashboard.json.
+
+A testnet dress rehearsal is only worth running if you can *assert* the outcomes, not
+eyeball them. During the rehearsal an operator drives a fixed set of scenarios — submit
+a good model, a broken repo, a freeze-cheat, a copy of the king — and this module reads
+the validator's published ``dashboard.json`` and confirms each one was handled the way
+the design says it should be.
+
+Every matcher is a pure function of the dashboard dict, so the assertion logic is fully
+unit-testable here and then pointed at a live validator during the rehearsal. Each
+returns a :class:`Scenario` with ``observed`` True/False and the evidence, so a partial
+run tells you exactly which behaviors have and haven't been exercised yet.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+
+
+@dataclass(frozen=True)
+class Scenario:
+    key: str
+    description: str
+    observed: bool
+    evidence: str = ""
+
+
+def _history(dash: dict) -> list[dict]:
+    return list(dash.get("history") or [])
+
+
+def _find(dash: dict, pred: Callable[[dict], bool]) -> Optional[dict]:
+    for row in _history(dash):
+        if pred(row):
+            return row
+    return None
+
+
+def _row_evidence(row: dict) -> str:
+    hk = str(row.get("hotkey") or "")
+    who = (hk[:10] + "…") if hk else "base"
+    return f"{row.get('model_repo', '?')} ({who}) at block {row.get('block', '?')}"
+
+
+def saw_crown(dash: dict) -> Scenario:
+    """A challenger genuinely beat the king and took the crown."""
+    row = _find(dash, lambda r: r.get("accepted") and r.get("verdict") == "challenger")
+    return Scenario("crown", "a challenger beat the king and was crowned",
+                    row is not None, _row_evidence(row) if row else "no crowning recorded")
+
+
+def saw_rejection(dash: dict) -> Scenario:
+    """A challenger was scored and lost fairly (lcb below delta) — the king held."""
+    row = _find(dash, lambda r: r.get("verdict") == "king" and not r.get("rejected_by")
+                and r.get("lcb") is not None)
+    return Scenario("rejection", "a challenger was scored and lost fairly (king held)",
+                    row is not None, _row_evidence(row) if row else "no fair rejection recorded")
+
+
+def saw_error_quarantine(dash: dict) -> Scenario:
+    """A broken/unloadable model produced an error row (and was quarantined)."""
+    row = _find(dash, lambda r: r.get("verdict") == "error")
+    ev = f"{_row_evidence(row)} — {row.get('error_reason', '?')}" if row else "no error row recorded"
+    return Scenario("error_quarantine", "a broken model was recorded as an error (not silently dropped)",
+                    row is not None, ev)
+
+
+def saw_freeze_cheat_rejected(dash: dict) -> Scenario:
+    """A model that beat the king only by holding the conditioning frame was rejected."""
+    row = _find(dash, lambda r: r.get("rejected_by") == "freeze_gate")
+    return Scenario("freeze_cheat", "a freeze-frame cheat was rejected by the gate",
+                    row is not None, _row_evidence(row) if row else "no freeze-gate rejection recorded")
+
+
+def saw_copy_rejected(dash: dict) -> Scenario:
+    """A copy of the king (exact or repackaged) was rejected without a full duel."""
+    row = _find(dash, lambda r: r.get("rejected_by") == "copy_of_king")
+    return Scenario("copy_of_king", "a copy of the king was rejected pre-duel",
+                    row is not None, _row_evidence(row) if row else "no copy-of-king rejection recorded")
+
+
+def saw_live_duel(dash: dict) -> Scenario:
+    """The dashboard surfaced at least one duel running on the GPU (the site moves
+    mid-duel). Reads `live_duels` (a list — a multi-eval-server validator can have
+    several duels in flight at once) and falls back to the older single-duel `live`
+    key for a dashboard snapshot published before live_duels existed."""
+    live_duels = dash.get("live_duels")
+    if live_duels is None:
+        single = dash.get("live")
+        live_duels = [single] if single else []
+    if not live_duels:
+        return Scenario("live_duel", "the dashboard shows a live in-flight duel", False,
+                        "no live duel at snapshot time")
+    first = live_duels[0]
+    ev = f"{first.get('model_repo')} dispatched at block {first.get('dispatched_block')}"
+    if len(live_duels) > 1:
+        ev += f" (+{len(live_duels) - 1} more in flight)"
+    return Scenario("live_duel", "the dashboard shows a live in-flight duel", True, ev)
+
+
+def saw_healthy(dash: dict) -> Scenario:
+    """The validator is NOT degraded (it is actually dueling, not burning)."""
+    degraded = dash.get("degraded")
+    return Scenario("healthy", "the validator is dueling, not degraded/burning",
+                    not degraded, f"degraded={degraded}" if degraded else "not degraded")
+
+
+#: The full rehearsal set, in the order an operator would naturally exercise them.
+ALL_SCENARIOS: tuple[Callable[[dict], Scenario], ...] = (
+    saw_healthy,
+    saw_crown,
+    saw_rejection,
+    saw_error_quarantine,
+    saw_copy_rejected,
+    saw_freeze_cheat_rejected,
+    saw_live_duel,
+)
+
+
+@dataclass
+class SmokeReport:
+    scenarios: list[Scenario] = field(default_factory=list)
+
+    @property
+    def passed(self) -> int:
+        return sum(1 for s in self.scenarios if s.observed)
+
+    @property
+    def total(self) -> int:
+        return len(self.scenarios)
+
+    @property
+    def complete(self) -> bool:
+        return all(s.observed for s in self.scenarios)
+
+    @property
+    def missing(self) -> list[Scenario]:
+        return [s for s in self.scenarios if not s.observed]
+
+
+def run_smoke(dash: dict, *, require_live: bool = False) -> SmokeReport:
+    """Evaluate every rehearsal scenario against one dashboard snapshot.
+
+    ``require_live`` includes the live-duel scenario in completeness — off by default
+    because whether a duel is running *at snapshot time* is timing-dependent, so it is
+    informational unless the operator is specifically checking it.
+    """
+    matchers = ALL_SCENARIOS if require_live else tuple(m for m in ALL_SCENARIOS if m is not saw_live_duel)
+    return SmokeReport([m(dash) for m in matchers])
+
+
+__all__ = ["Scenario", "SmokeReport", "run_smoke", "ALL_SCENARIOS"]
diff --git a/leoma/delivery/commands.py b/leoma/delivery/commands.py
index ba06b49..aac4468 100644
--- a/leoma/delivery/commands.py
+++ b/leoma/delivery/commands.py
@@ -70,6 +70,144 @@ def start_eval_server():
     eval_main()
 
 
+@cli.command()
+def preflight():
+    """Check whether this validator is ready to launch — a hard gate, not a hope.
+
+    Verifies every pin the subnet needs before it will crown anyone: the genesis
+    king, the corpus manifest, the consensus surface, and (if configured) that the
+    eval box is on the same chain.toml + scoring code. Exits non-zero if anything is
+    a hard FAIL, so it can gate a launch script.
+    """
+    import os
+
+    from leoma.bootstrap import emit_log as log, emit_header as log_header
+    from leoma.app.preflight import PASS, WARN, FAIL, run_preflight
+    from leoma.eval.codehash import eval_code_digest
+    from leoma.infra.chain_config import CONSENSUS_DIGEST, SEED_DIGEST, SPEC
+
+    log_header("Leoma Preflight")
+
+    # --- optional I/O the pure checks consume ---
+    corpus_fetched_digest = None
+    corpus_error = None
+    if SPEC.corpus.pinned:
+        try:
+            from leoma.eval.dataset import fetch_manifest
+            from leoma.infra.storage_backend import create_source_read_client
+
+            manifest = fetch_manifest(create_source_read_client(), SPEC.corpus)
+            corpus_fetched_digest = manifest.source_digest
+        except Exception as e:  # noqa: BLE001 — a fetch failure is a WARN, not a crash
+            corpus_error = str(e)
+
+    # EVAL_SERVER_URLS (plural) lets an operator configure several independently-run
+    # eval-server processes; checking only EVAL_SERVER_URL here would silently skip
+    # every server but the first. _parse_eval_server_urls is the same pure parser the
+    # validator itself uses, so the two never drift apart.
+    eval_url_env = os.environ.get("EVAL_SERVER_URL", "")
+    eval_urls_env = os.environ.get("EVAL_SERVER_URLS", "")
+    eval_probes = ()
+    if eval_url_env or eval_urls_env.strip():
+        import httpx
+
+        from leoma.app.preflight import EvalServerProbe
+        from leoma.app.validator.main import _parse_eval_server_urls
+
+        urls = _parse_eval_server_urls(eval_urls_env, eval_url_env or "http://localhost:9000")
+        probes = []
+        for url in urls:
+            try:
+                health = httpx.get(f"{url}/health", timeout=httpx.Timeout(15.0)).json()
+                probes.append(EvalServerProbe(url, health, None))
+            except Exception as e:  # noqa: BLE001
+                probes.append(EvalServerProbe(url, None, str(e)))
+        eval_probes = tuple(probes)
+
+    report = run_preflight(
+        seed_digest=SEED_DIGEST,
+        corpus_pinned=SPEC.corpus.pinned,
+        manifest_digest=SPEC.corpus.manifest_digest,
+        consensus_digest=CONSENSUS_DIGEST,
+        eval_code_digest=eval_code_digest(),
+        own_bucket=os.environ.get("R2_OWN_BUCKET"),
+        wallet_name=os.environ.get("WALLET_NAME"),
+        hotkey_name=os.environ.get("HOTKEY_NAME"),
+        corpus_fetched_digest=corpus_fetched_digest,
+        corpus_error=corpus_error,
+        eval_servers=eval_probes,
+    )
+
+    icon = {PASS: "✓", WARN: "▲", FAIL: "✗"}
+    level = {PASS: "success", WARN: "warn", FAIL: "error"}
+    for c in report.checks:
+        log(f"{icon[c.status]} {c.name}: {c.detail}", level[c.status])
+
+    log_header("Ready" if report.ready else "NOT READY")
+    if report.ready:
+        msg = "all checks pass" if not report.warnings else f"ready ({len(report.warnings)} warning(s) — review above)"
+        log(msg, "success")
+    else:
+        log(f"{len(report.failures)} blocking failure(s) — the validator will burn to UID 0 until fixed", "error")
+        raise SystemExit(1)
+
+
+@cli.command()
+@click.argument("source", default="")
+@click.option("--require-live", is_flag=True, help="Also require an in-flight duel at snapshot time")
+def smoke(source, require_live):
+    """Confirm a testnet run actually exercised each expected outcome.
+
+    Reads a validator's published dashboard.json (a URL, a file path, or
+    LEOMA_DASHBOARD_URL) and checks that the rehearsal scenarios were observed: a
+    genuine crown, a fair rejection, a quarantined broken model, a copy-of-king
+    rejection, a freeze-cheat rejection. Reports exactly which behaviors have and
+    haven't been exercised yet, so you can drive the missing ones.
+    """
+    import json
+    import os
+
+    from leoma.bootstrap import emit_log as log, emit_header as log_header
+    from leoma.app.smoke import run_smoke
+
+    src = source or os.environ.get("LEOMA_DASHBOARD_URL", "")
+    if not src:
+        raise click.ClickException("give a dashboard.json URL or path (or set LEOMA_DASHBOARD_URL)")
+
+    log_header("Leoma Smoke — testnet scenarios")
+    if src.startswith("http://") or src.startswith("https://"):
+        import httpx
+
+        try:
+            resp = httpx.get(src, timeout=httpx.Timeout(20.0))
+            resp.raise_for_status()
+            dash = resp.json()
+        except httpx.HTTPError as e:
+            raise click.ClickException(f"could not fetch dashboard from {src}: {e}")
+        except ValueError as e:  # json.JSONDecodeError subclasses ValueError
+            raise click.ClickException(f"{src} did not return valid JSON: {e}")
+    else:
+        try:
+            with open(src) as f:
+                dash = json.load(f)
+        except OSError as e:
+            raise click.ClickException(f"could not read dashboard file {src}: {e}")
+        except json.JSONDecodeError as e:
+            raise click.ClickException(f"{src} is not valid JSON: {e}")
+
+    report = run_smoke(dash, require_live=require_live)
+    for s in report.scenarios:
+        log(f"{'✓' if s.observed else '·'} {s.description} — {s.evidence}",
+            "success" if s.observed else "warn")
+
+    log_header(f"{report.passed}/{report.total} scenarios observed")
+    if report.complete:
+        log("every rehearsal scenario was exercised and handled correctly", "success")
+    else:
+        log("still to drive: " + ", ".join(s.key for s in report.missing), "warn")
+        raise SystemExit(1)
+
+
 @cli.group()
 def corpus():
     """Video corpus management commands.
diff --git a/tests/unit/test_cli_smoke.py b/tests/unit/test_cli_smoke.py
new file mode 100644
index 0000000..389d404
--- /dev/null
+++ b/tests/unit/test_cli_smoke.py
@@ -0,0 +1,98 @@
+"""CLI-level crash handling for `leoma smoke`.
+
+Before this, an unreachable URL or malformed JSON crashed with a raw traceback
+(httpx.ConnectError / json.JSONDecodeError bubbling straight out of the command) instead
+of a clean, actionable CLI error — the exact opposite of what an operator needs while
+driving a testnet rehearsal. These tests pin that both failure modes, for both the URL
+and file-path sources, now exit cleanly with a readable message.
+"""
+import json
+
+import httpx
+import pytest
+from click.testing import CliRunner
+
+from leoma.delivery.commands import cli
+
+
+@pytest.fixture
+def runner():
+    return CliRunner()
+
+
+def _good_dashboard():
+    return {
+        "history": [{"accepted": True, "verdict": "challenger", "hotkey": "5win",
+                     "model_repo": "u/leoma-win", "block": 100}],
+        "live_duels": [],
+        "degraded": None,
+    }
+
+
+class TestSmokeFromFile:
+    def test_a_valid_dashboard_file_runs_cleanly(self, runner, tmp_path):
+        path = tmp_path / "dashboard.json"
+        path.write_text(json.dumps(_good_dashboard()))
+        result = runner.invoke(cli, ["smoke", str(path)])
+        assert "scenarios observed" in result.output
+
+    def test_a_missing_file_exits_cleanly_not_a_traceback(self, runner, tmp_path):
+        missing = tmp_path / "does_not_exist.json"
+        result = runner.invoke(cli, ["smoke", str(missing)])
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert "could not read dashboard file" in result.output
+
+    def test_malformed_json_exits_cleanly_not_a_traceback(self, runner, tmp_path):
+        path = tmp_path / "dashboard.json"
+        path.write_text("{not valid json")
+        result = runner.invoke(cli, ["smoke", str(path)])
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert "not valid JSON" in result.output
+
+
+class TestSmokeFromUrl:
+    def test_an_unreachable_url_exits_cleanly_not_a_traceback(self, runner, monkeypatch):
+        def _raise(*args, **kwargs):
+            raise httpx.ConnectError("connection refused")
+
+        monkeypatch.setattr(httpx, "get", _raise)
+        result = runner.invoke(cli, ["smoke", "http://nonexistent-eval-box:9000/dashboard.json"])
+        assert result.exit_code != 0
+        assert result.exception is None or isinstance(result.exception, SystemExit)
+        assert "could not fetch dashboard" in result.output
+
+    def test_a_non_200_response_exits_cleanly(self, runner, monkeypatch):
+        def _fake_get(url, timeout=None):
+            return httpx.Response(500, request=httpx.Request("GET", url))
+
+        monkeypatch.setattr(httpx, "get", _fake_get)
+        result = runner.invoke(cli, ["smoke", "http://eval-box:9000/dashboard.json"])
+        assert result.exit_code != 0
+        assert "could not fetch dashboard" in result.output
+
+    def test_a_valid_url_response_runs_cleanly(self, runner, monkeypatch):
+        def _fake_get(url, timeout=None):
+            return httpx.Response(200, json=_good_dashboard(), request=httpx.Request("GET", url))
+
+        monkeypatch.setattr(httpx, "get", _fake_get)
+        result = runner.invoke(cli, ["smoke", "http://eval-box:9000/dashboard.json"])
+        assert "scenarios observed" in result.output
+
+    def test_non_json_body_exits_cleanly(self, runner, monkeypatch):
+        def _fake_get(url, timeout=None):
+            return httpx.Response(200, content=b"not json", request=httpx.Request("GET", url))
+
+        monkeypatch.setattr(httpx, "get", _fake_get)
+        result = runner.invoke(cli, ["smoke", "http://eval-box:9000/dashboard.json"])
+        assert result.exit_code != 0
+        assert "did not return valid JSON" in result.output
+
+
+class TestSmokeMissingSource:
+    def test_no_source_and_no_env_var_is_a_clean_usage_error(self, runner, monkeypatch):
+        monkeypatch.delenv("LEOMA_DASHBOARD_URL", raising=False)
+        result = runner.invoke(cli, ["smoke"])
+        assert result.exit_code != 0
+        assert "dashboard.json URL or path" in result.output
diff --git a/tests/unit/test_preflight.py b/tests/unit/test_preflight.py
new file mode 100644
index 0000000..7bfe7c0
--- /dev/null
+++ b/tests/unit/test_preflight.py
@@ -0,0 +1,189 @@
+"""The launch readiness gate.
+
+Every pin the subnet needs fails safe-but-silent: an unpinned seed, an unpinned
+corpus, or a stale eval box all make the validator burn to UID 0 rather than crown.
+Preflight is the one place that says WHY before an operator flips the switch. These
+tests pin the classification of each check and the overall ready/not-ready verdict.
+"""
+
+from leoma.app.preflight import (
+    FAIL,
+    PASS,
+    WARN,
+    EvalServerProbe,
+    check_corpus_reachable,
+    check_eval_server,
+    check_eval_servers,
+    check_seed,
+    run_preflight,
+)
+
+GOOD = dict(
+    seed_digest="sha256:" + "5" * 64,
+    corpus_pinned=True,
+    manifest_digest="sha256:" + "a" * 64,
+    consensus_digest="sha256:" + "c" * 64,
+    eval_code_digest="sha256:" + "e" * 64,
+    own_bucket="leoma-state",
+    wallet_name="val",
+    hotkey_name="hk",
+)
+
+
+def _good_probe():
+    return EvalServerProbe(
+        "http://eval:9000",
+        {"consensus_digest": GOOD["consensus_digest"], "eval_code_digest": GOOD["eval_code_digest"]},
+        None,
+    )
+
+
+class TestOverallVerdict:
+    def test_a_fully_configured_validator_is_ready(self):
+        r = run_preflight(**GOOD, corpus_fetched_digest=GOOD["manifest_digest"],
+                          eval_servers=(_good_probe(),))
+        assert r.ready
+        assert r.failures == ()
+
+    def test_an_unpinned_seed_blocks_launch(self):
+        r = run_preflight(**{**GOOD, "seed_digest": ""})
+        assert not r.ready
+        assert any(c.name == "seed_digest" and c.status == FAIL for c in r.checks)
+
+    def test_an_unpinned_corpus_blocks_launch(self):
+        r = run_preflight(**{**GOOD, "corpus_pinned": False, "manifest_digest": ""})
+        assert not r.ready
+        assert any(c.name == "corpus_pin" and c.status == FAIL for c in r.checks)
+
+    def test_a_missing_state_bucket_blocks_launch(self):
+        r = run_preflight(**{**GOOD, "own_bucket": None})
+        assert not r.ready
+
+    def test_warnings_do_not_block_launch(self):
+        # No eval server configured and no corpus fetch -> warnings only, still ready.
+        r = run_preflight(**GOOD)
+        assert r.ready
+        assert len(r.warnings) >= 1  # eval_server + corpus_fetch are warns here
+
+
+class TestCorpusReachable:
+    def test_matching_digest_passes(self):
+        c = check_corpus_reachable("sha256:aaa", "sha256:aaa", None)
+        assert c.status == PASS
+
+    def test_a_drifted_manifest_is_a_hard_fail(self):
+        """The bucket serves a manifest that isn't the pinned one — validators would
+        grade different exams. Must block."""
+        c = check_corpus_reachable("sha256:bbb", "sha256:aaa", None)
+        assert c.status == FAIL
+        assert "does NOT match" in c.detail
+
+    def test_a_fetch_error_is_only_a_warning(self):
+        c = check_corpus_reachable(None, "sha256:aaa", "connection refused")
+        assert c.status == WARN
+
+    def test_no_credentials_is_a_warning_not_a_block(self):
+        c = check_corpus_reachable(None, "sha256:aaa", None)
+        assert c.status == WARN
+
+
+class TestEvalServer:
+    def test_matching_box_passes(self):
+        c = check_eval_server({"consensus_digest": "sha256:c", "eval_code_digest": "sha256:e"},
+                              "sha256:c", "sha256:e")
+        assert c.status == PASS
+
+    def test_a_stale_consensus_surface_is_a_hard_fail(self):
+        """The single most likely consensus failure — an operator who redeployed some
+        boxes but not others. Must block."""
+        c = check_eval_server({"consensus_digest": "sha256:OLD", "eval_code_digest": "sha256:e"},
+                              "sha256:NEW", "sha256:e")
+        assert c.status == FAIL
+        assert "DIFFERENT consensus surface" in c.detail
+
+    def test_stale_scoring_code_is_a_hard_fail(self):
+        c = check_eval_server({"consensus_digest": "sha256:c", "eval_code_digest": "sha256:OLD"},
+                              "sha256:c", "sha256:NEW")
+        assert c.status == FAIL
+        assert "DIFFERENT scoring code" in c.detail
+
+    def test_no_eval_server_configured_is_a_warning(self):
+        c = check_eval_server(None, "sha256:c", "sha256:e")
+        assert c.status == WARN
+
+    def test_unreachable_eval_server_is_a_warning(self):
+        c = check_eval_server(None, "sha256:c", "sha256:e", error="timeout")
+        assert c.status == WARN
+
+    def test_missing_eval_code_digest_is_a_warning_not_a_silent_pass(self):
+        """A box whose /health doesn't report eval_code_digest at all (a build old
+        enough to predate the field) gives us zero evidence its scoring code matches
+        — this must surface, not silently PASS."""
+        c = check_eval_server({"consensus_digest": "sha256:c"}, "sha256:c", "sha256:e")
+        assert c.status == WARN
+        assert "could not be verified" in c.detail
+
+    def test_a_custom_name_labels_the_check(self):
+        c = check_eval_server({"consensus_digest": "sha256:c", "eval_code_digest": "sha256:e"},
+                              "sha256:c", "sha256:e", name="eval_server[http://a:9000]")
+        assert c.name == "eval_server[http://a:9000]"
+
+
+class TestSeedDigestFormat:
+    def test_a_valid_hippius_digest_passes(self):
+        assert check_seed("sha256:" + "a" * 64).status == PASS
+
+    def test_a_valid_hf_commit_sha_passes(self):
+        assert check_seed("hf:" + "a" * 40).status == PASS
+
+    def test_blank_fails(self):
+        c = check_seed("")
+        assert c.status == FAIL
+        assert "is empty" in c.detail
+
+    def test_malformed_digest_fails(self):
+        """A truncated/mistyped pin would otherwise resolve to nothing at genesis
+        time — catch it here, not mid-launch."""
+        c = check_seed("sha256:tooshort")
+        assert c.status == FAIL
+        assert "not a recognized digest" in c.detail
+
+    def test_wrong_prefix_fails(self):
+        c = check_seed("md5:" + "a" * 32)
+        assert c.status == FAIL
+
+
+class TestEvalServersFleet:
+    def test_no_probes_gives_one_warning_check(self):
+        checks = check_eval_servers((), "sha256:c", "sha256:e")
+        assert len(checks) == 1
+        assert checks[0].name == "eval_server"
+        assert checks[0].status == WARN
+
+    def test_a_single_probe_keeps_the_unlabeled_name(self):
+        """Single-server operators shouldn't see a URL-qualified name change under them."""
+        probe = EvalServerProbe("http://only:9000", {"consensus_digest": "sha256:c",
+                                                     "eval_code_digest": "sha256:e"}, None)
+        checks = check_eval_servers((probe,), "sha256:c", "sha256:e")
+        assert len(checks) == 1
+        assert checks[0].name == "eval_server"
+        assert checks[0].status == PASS
+
+    def test_several_probes_are_each_checked_and_labeled_by_url(self):
+        """A stale box among several configured servers must not hide behind a
+        healthy sibling — every configured URL gets checked independently."""
+        healthy = EvalServerProbe("http://a:9000", {"consensus_digest": "sha256:c",
+                                                     "eval_code_digest": "sha256:e"}, None)
+        stale = EvalServerProbe("http://b:9000", {"consensus_digest": "sha256:OLD",
+                                                   "eval_code_digest": "sha256:e"}, None)
+        checks = check_eval_servers((healthy, stale), "sha256:c", "sha256:e")
+        assert len(checks) == 2
+        by_name = {c.name: c for c in checks}
+        assert by_name["eval_server[http://a:9000]"].status == PASS
+        assert by_name["eval_server[http://b:9000]"].status == FAIL
+
+    def test_every_server_stale_fails_the_whole_fleet_check(self):
+        stale_a = EvalServerProbe("http://a:9000", {"consensus_digest": "sha256:OLD"}, None)
+        stale_b = EvalServerProbe("http://b:9000", None, "connection refused")
+        checks = check_eval_servers((stale_a, stale_b), "sha256:c", "sha256:e")
+        assert all(c.status != PASS for c in checks)
diff --git a/tests/unit/test_smoke.py b/tests/unit/test_smoke.py
new file mode 100644
index 0000000..69ea4e8
--- /dev/null
+++ b/tests/unit/test_smoke.py
@@ -0,0 +1,134 @@
+"""Testnet scenario matchers.
+
+A dress rehearsal is only worth running if the outcomes can be asserted. These
+matchers read a validator's dashboard.json and confirm each expected behavior
+happened; the tests pin what counts as "observed" for each scenario so a green smoke
+run genuinely means the subnet did the right thing.
+"""
+
+from leoma.app.smoke import (
+    run_smoke,
+    saw_copy_rejected,
+    saw_crown,
+    saw_error_quarantine,
+    saw_freeze_cheat_rejected,
+    saw_healthy,
+    saw_live_duel,
+    saw_rejection,
+)
+
+
+def _dash(history=None, live=None, degraded=None):
+    return {"history": history or [], "live": live, "degraded": degraded}
+
+
+def _crown_row():
+    return {"accepted": True, "verdict": "challenger", "hotkey": "5win", "model_repo": "u/leoma-win", "block": 100}
+
+
+def _held_row():
+    return {"accepted": False, "verdict": "king", "lcb": -0.01, "hotkey": "5lose",
+            "model_repo": "u/leoma-lose", "block": 101}
+
+
+class TestIndividualScenarios:
+    def test_crown_needs_an_accepted_challenger(self):
+        assert saw_crown(_dash([_crown_row()])).observed
+        assert not saw_crown(_dash([_held_row()])).observed
+
+    def test_fair_rejection_needs_a_scored_loss_not_a_gate_rejection(self):
+        assert saw_rejection(_dash([_held_row()])).observed
+        # a gate rejection (rejected_by set) is NOT a fair scored loss
+        gate = {"verdict": "king", "rejected_by": "freeze_gate", "lcb": 0.004}
+        assert not saw_rejection(_dash([gate])).observed
+
+    def test_error_quarantine_needs_an_error_row(self):
+        row = {"verdict": "error", "error_reason": "model_not_found", "hotkey": "5bad", "model_repo": "u/x"}
+        s = saw_error_quarantine(_dash([row]))
+        assert s.observed and "model_not_found" in s.evidence
+
+    def test_freeze_cheat_needs_a_freeze_gate_rejection(self):
+        row = {"verdict": "king", "rejected_by": "freeze_gate"}
+        assert saw_freeze_cheat_rejected(_dash([row])).observed
+        assert not saw_freeze_cheat_rejected(_dash([_held_row()])).observed
+
+    def test_copy_needs_a_copy_of_king_rejection(self):
+        row = {"verdict": "king", "rejected_by": "copy_of_king"}
+        assert saw_copy_rejected(_dash([row])).observed
+
+    def test_live_duel_needs_a_live_block(self):
+        assert saw_live_duel(_dash(live={"model_repo": "u/x", "dispatched_block": 9})).observed
+        assert not saw_live_duel(_dash()).observed
+
+    def test_live_duel_prefers_the_live_duels_list(self):
+        """A multi-eval-server validator publishes live_duels (a list) instead of the
+        older single live dict — the smoke tool must read the current key, not just
+        the back-compat one."""
+        dash = {"history": [], "live_duels": [{"model_repo": "u/multi", "dispatched_block": 42}],
+                "degraded": None}
+        s = saw_live_duel(dash)
+        assert s.observed
+        assert "u/multi" in s.evidence
+
+    def test_live_duel_reports_how_many_more_are_in_flight(self):
+        dash = {"history": [], "live_duels": [
+            {"model_repo": "u/a", "dispatched_block": 1},
+            {"model_repo": "u/b", "dispatched_block": 2},
+        ], "degraded": None}
+        s = saw_live_duel(dash)
+        assert s.observed
+        assert "+1 more" in s.evidence
+
+    def test_live_duel_falls_back_to_live_when_live_duels_key_is_absent(self):
+        """A dashboard snapshot published before live_duels existed still works."""
+        dash = {"history": [], "live": {"model_repo": "u/old", "dispatched_block": 5}, "degraded": None}
+        assert saw_live_duel(dash).observed
+
+    def test_live_duel_is_not_observed_when_live_duels_is_an_empty_list(self):
+        dash = {"history": [], "live_duels": [], "degraded": None}
+        assert not saw_live_duel(dash).observed
+
+    def test_healthy_is_the_absence_of_a_degraded_reason(self):
+        assert saw_healthy(_dash()).observed
+        assert not saw_healthy(_dash(degraded="corpus_unpinned")).observed
+
+
+class TestFullReport:
+    def _rich(self):
+        return _dash(
+            history=[
+                _crown_row(),
+                _held_row(),
+                {"verdict": "error", "error_reason": "model_not_found", "model_repo": "u/e"},
+                {"verdict": "king", "rejected_by": "copy_of_king", "model_repo": "u/c"},
+                {"verdict": "king", "rejected_by": "freeze_gate", "model_repo": "u/f"},
+            ],
+            live={"model_repo": "u/live", "dispatched_block": 200},
+        )
+
+    def test_a_full_rehearsal_is_complete(self):
+        r = run_smoke(self._rich(), require_live=True)
+        assert r.complete
+        assert r.passed == r.total
+
+    def test_a_partial_run_reports_exactly_what_is_missing(self):
+        # only a crown so far
+        r = run_smoke(_dash([_crown_row()]))
+        assert not r.complete
+        missing = {s.key for s in r.missing}
+        assert "copy_of_king" in missing
+        assert "freeze_cheat" in missing
+        assert "error_quarantine" in missing
+        assert "crown" not in missing
+
+    def test_live_is_excluded_from_completeness_by_default(self):
+        """Whether a duel is running at snapshot time is timing-dependent, so it is
+        informational unless explicitly required."""
+        r = run_smoke(self._rich())  # require_live=False
+        assert all(s.key != "live_duel" for s in r.scenarios)
+        assert r.complete
+
+    def test_a_degraded_validator_fails_the_healthy_scenario(self):
+        r = run_smoke(_dash([_crown_row()], degraded="no_seed_digest"))
+        assert not r.complete
+        assert any(s.key == "healthy" and not s.observed for s in r.scenarios)