From 2658f737e300978e413a720809577bcc66b8b0fd Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:36:14 +0000 Subject: [PATCH 1/2] fix(seal): stop epoch desync between design emit and real-seal Design re-read the chain schedule inside emit_leaves at late tempo, so an epoch-boundary flip could relabel the set as E+1 and permanently skip E. Real-seal only tried the current epoch, leaving /weights/latest stuck on an old chain-scale bundle. Catch up skipped epochs, widen the late window, walk back sealable epochs in prod-real-seal, and add validator pressure-verify logs when the sealed epoch lags the chain. --- crates/design-challenge/src/lib.rs | 31 +++++- crates/design-challenge/src/orchestrator.rs | 105 ++++++++++++++++--- crates/validator-verify/src/coordination.rs | 3 + crates/validator/src/epoch_loop.rs | 73 +++++++++++++ deploy/AGENTS.md | 2 +- deploy/scripts/prod-real-seal.sh | 110 ++++++++++++++++---- 6 files changed, 283 insertions(+), 41 deletions(-) diff --git a/crates/design-challenge/src/lib.rs b/crates/design-challenge/src/lib.rs index a7101a170..e117631f9 100644 --- a/crates/design-challenge/src/lib.rs +++ b/crates/design-challenge/src/lib.rs @@ -44,7 +44,10 @@ pub use design_store_pg::DbDesignStore; pub use host_sim::{ force_sim_refusal_reason, host_sim_allowed, is_prod_env, require_host_sim_for_force, }; -pub use orchestrator::{ErrorClass, Orchestrator, OrchestratorConfig}; +pub use orchestrator::{ + design_emit_plan, DesignEmitPlan, ErrorClass, Orchestrator, OrchestratorConfig, + DESIGN_EMIT_LATE_BLOCKS, +}; /// Crate identity smoke. #[must_use] @@ -62,4 +65,30 @@ mod tests { assert_eq!(CHALLENGE_ID, "design"); assert_eq!(SCORING_VERSION, 3); } + + #[test] + fn emit_plan_waits_until_late_tempo_for_current_epoch() { + assert!(design_emit_plan(10, 11, 200, 360, 1000).is_none()); + let p = design_emit_plan(10, 11, 360 - DESIGN_EMIT_LATE_BLOCKS, 360, 1000).unwrap(); + assert_eq!( + p, + DesignEmitPlan { + epoch: 11, + pin_block: 1000 + } + ); + } + + #[test] + fn emit_plan_catches_up_skipped_epochs_without_waiting() { + // Prod failure mode: award/boundary race skipped 24413 while chain is 24423. + let p = design_emit_plan(24412, 24423, 50, 360, 8_815_687).unwrap(); + assert_eq!(p.epoch, 24413); + assert_eq!(p.pin_block, 8_815_687 - (24423 - 24413) * 360); + } + + #[test] + fn emit_plan_noop_when_already_emitted_current() { + assert!(design_emit_plan(11, 11, 350, 360, 1000).is_none()); + } } diff --git a/crates/design-challenge/src/orchestrator.rs b/crates/design-challenge/src/orchestrator.rs index c7d11f38f..085621362 100644 --- a/crates/design-challenge/src/orchestrator.rs +++ b/crates/design-challenge/src/orchestrator.rs @@ -285,8 +285,8 @@ impl Orchestrator { } } - /// Late-tempo D24 filler: `NotAttempted` coverage when no admin award fired - /// (waits ~last 48 blocks so `award_round` can land Score leaves first). + /// D24 filler + catch-up: covers epochs with no admin award, and repairs + /// gaps left by the end-of-epoch boundary race (see [`design_emit_plan`]). pub async fn run_emitter(self: Arc) where C: Sync, @@ -310,14 +310,21 @@ impl Orchestrator { let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) .map_err(|e| format!("schedule: {e}"))?; let epoch = state.subnet_epoch_index; - if epoch == 0 || self.emitted_epoch.load(Ordering::Relaxed) >= epoch { - return Ok(false); - } let tempo = u64::from(state.tempo.max(1)); - if state.blocks_since_last_step.saturating_add(48) < tempo { + let last = self.emitted_epoch.load(Ordering::Relaxed); + let Some(plan) = design_emit_plan( + last, + epoch, + state.blocks_since_last_step, + tempo, + state.last_epoch_block, + ) else { return Ok(false); - } - self.emit_leaves().await?; + }; + // Pin epoch + block from this tick's schedule snapshot — do **not** + // re-read chain inside emit (end-of-epoch flip used to relabel the set + // as E+1 and permanently skip E, starving real-seal with D24 409s). + self.emit_leaves_at(plan.epoch, plan.pin_block).await?; Ok(true) } @@ -1089,15 +1096,22 @@ impl Orchestrator { async fn emit_leaves(&self) -> Result<(), String> { let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) .map_err(|e| format!("schedule: {e}"))?; - // Label with the *current* chain epoch (see the prism emitter for the - // full rationale): the expected set is pinned at `last_epoch_block`, - // so the label and the covered metagraph must refer to the same epoch - // or D24 exact-match against the other challenge 409s under churn. - let epoch = state.subnet_epoch_index; + self.emit_leaves_at(state.subnet_epoch_index, state.last_epoch_block) + .await + } + + /// Submit a D24-complete design leaf set for a pinned `(epoch, pin_block)`. + /// + /// `pin_block` must be that epoch's start block (`LastEpochBlock` while the + /// epoch is current, or `current_last_epoch_block - k*tempo` when catching up). + async fn emit_leaves_at(&self, epoch: u64, pin_block: u64) -> Result<(), String> { + if epoch == 0 { + return Err("refuse emit for epoch 0".into()); + } let block_hash = self .chain - .block_hash(state.last_epoch_block) - .map_err(|e| format!("block_hash: {e}"))?; + .block_hash(pin_block) + .map_err(|e| format!("block_hash@{pin_block}: {e}"))?; let expected: ExpectedSet = expected_set_at_chain( &trustroot::ParticipantPolicy::AllMetagraphHotkeys, PinnedBlockHash::new(block_hash), @@ -1131,17 +1145,74 @@ impl Orchestrator { submit_signed_leaf_set(self.gateway.as_ref(), &signed) .await .map_err(|e| e.to_string())?; - self.emitted_epoch.store(epoch, Ordering::Relaxed); + self.emitted_epoch.fetch_max(epoch, Ordering::Relaxed); info!( epoch, participants = expected_set.len(), - last_epoch_block = state.last_epoch_block, + pin_block, "design leaf set submitted" ); Ok(()) } } +/// How many blocks before epoch end the NotAttempted filler may run. +/// +/// Wider than the historical 48-block window so `base-real-seal` (10 min) still +/// has time to seal after design emits, while leaving most of the epoch for +/// `award_round` to land Score leaves first (first-write-wins). +pub const DESIGN_EMIT_LATE_BLOCKS: u64 = 96; + +/// Planned design leaf emission for one emitter tick. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DesignEmitPlan { + /// Epoch label for the leaf set. + pub epoch: u64, + /// Metagraph pin block (epoch start). + pub pin_block: u64, +} + +/// Decide whether/which epoch the design filler should emit. +/// +/// - Catch up `last_emitted+1` when behind by more than one epoch (repairs the +/// end-of-epoch relabel race that skipped alternate epochs in prod). +/// - Otherwise wait until the last [`DESIGN_EMIT_LATE_BLOCKS`] of the current +/// epoch so admin awards can submit Score leaves first. +#[must_use] +pub fn design_emit_plan( + last_emitted: u64, + current_epoch: u64, + blocks_since_last_step: u64, + tempo: u64, + current_last_epoch_block: u64, +) -> Option { + if current_epoch == 0 { + return None; + } + let tempo = tempo.max(1); + if last_emitted >= current_epoch { + return None; + } + // Sequential catch-up for skipped epochs (award path / boundary race). + if last_emitted + 1 < current_epoch { + let target = last_emitted + 1; + let epochs_back = current_epoch.saturating_sub(target); + let pin_block = current_last_epoch_block.saturating_sub(epochs_back.saturating_mul(tempo)); + return Some(DesignEmitPlan { + epoch: target, + pin_block, + }); + } + // Current epoch: late-tempo filler only. + if blocks_since_last_step.saturating_add(DESIGN_EMIT_LATE_BLOCKS) < tempo { + return None; + } + Some(DesignEmitPlan { + epoch: current_epoch, + pin_block: current_last_epoch_block, + }) +} + #[async_trait] impl AdminAwardHook for Orchestrator { async fn on_winners(&self, round_id: u64, _harness_ids: &[String]) -> Result<(), String> { diff --git a/crates/validator-verify/src/coordination.rs b/crates/validator-verify/src/coordination.rs index f7b590ad1..e532db1b7 100644 --- a/crates/validator-verify/src/coordination.rs +++ b/crates/validator-verify/src/coordination.rs @@ -276,6 +276,9 @@ pub struct WeightsLatestView { /// IP) fails loudly instead of surfacing as an opaque bundle verify error. #[serde(default)] pub netuid: Option, + /// Metagraph block pinned by the seal (for prune / lag pressure checks). + #[serde(default)] + pub metagraph_block: Option, } impl WeightsLatestView { diff --git a/crates/validator/src/epoch_loop.rs b/crates/validator/src/epoch_loop.rs index f733629b6..821f72afd 100644 --- a/crates/validator/src/epoch_loop.rs +++ b/crates/validator/src/epoch_loop.rs @@ -270,6 +270,39 @@ where // Fail-closed burn (sealed=false) or incomplete legacy body — no Match path. return Ok(None); }; + // Pressure / verify: sealed vector must stay near the live chain epoch. + // A stuck real-seal (D24 incomplete) leaves `/v1/weights/latest` on an old + // chain-scale bundle; Match can still succeed while emission is stale. + if let Some(cfg) = submit { + if let Ok(chain_epoch) = chain.subnet_epoch_index(cfg.netuid) { + let lag = chain_epoch.saturating_sub(epoch); + if lag > 1 { + warn!( + event = "validator_seal_lag", + sealed_epoch = epoch, + chain_epoch, + lag_epochs = lag, + metagraph_block = ?latest.metagraph_block, + "pressure verify: sealed weights lag chain epoch; check design/prism emit + base-real-seal" + ); + } + } + if let (Ok(tip), Some(mg_block)) = (chain.current_block(), latest.metagraph_block) { + let block_lag = tip.saturating_sub(mg_block); + // Public Finney RPC prunes ~256 blocks; beyond that Match cannot + // re-fetch the seal's metagraph on a cold RPC (ops red line). + if block_lag > 256 { + warn!( + event = "validator_seal_metagraph_stale", + sealed_epoch = epoch, + metagraph_block = mg_block, + tip_block = tip, + block_lag, + "pressure verify: seal metagraph_block outside ~256-block prune window" + ); + } + } + } let outcome = fetch_and_compare(client, epoch, chain, trust).await; match &outcome { ComparisonOutcome::Match { @@ -708,4 +741,44 @@ mod tests { maybe_submit_match(&outcome, &chain, &ReadyDrand, None, &dedupe); assert!(chain.call_log().is_empty()); } + + #[tokio::test] + async fn tick_pressure_verify_allows_match_when_seal_lags_chain() { + // Same metagraph as the sealed fixture, but chain epoch/tip far ahead — + // pressure-verify warns (validator_seal_lag) yet Match must still proceed. + let epoch = 77u64; + let (client, _chain, trust, merkle_root, _) = sealed_match_fixture(epoch).await; + let miner = [0xA1u8; 32]; + let chain = FakeChain::new(FakeChainConfig { + current_block: 10_000, + subnet_epoch_index: epoch + 11, + hotkeys: vec![miner.to_vec()], + owner_hotkey: miner.to_vec(), + commit_reveal_enabled: true, + last_epoch_block: 500, + ..FakeChainConfig::default() + }); + let dedupe = EpochSubmitDedupe::new(); + let submit = CoordinationSubmitConfig { + netuid: 1, + hotkey: vec![0xBBu8; 32], + version_key: 3, + epoch_length: 360, + }; + let out = coordination_compare_once(&client, &chain, &trust, Some(&submit), &dedupe) + .await + .expect("ok") + .expect("some"); + match out { + ComparisonOutcome::Match { + epoch: e, + merkle_root: root, + .. + } => { + assert_eq!(e, epoch); + assert_eq!(root, merkle_root); + } + other => panic!("expected Match despite seal lag, got {other:?}"), + } + } } diff --git a/deploy/AGENTS.md b/deploy/AGENTS.md index 66335f688..6e6188a92 100644 --- a/deploy/AGENTS.md +++ b/deploy/AGENTS.md @@ -96,7 +96,7 @@ cargo run -q --release -p weights-smoke -- \ A seal older than ~256 blocks can never be verified by the validator (public RPC prunes state) — if `GET /v1/weights/latest` shows `metagraph_block` lagging tip by thousands of blocks, check `systemctl status base-burn-seal.timer` and `/var/log/base-burn-seal.log` on the master. -**Real-epoch sealer (post burn-seal retirement):** `base-real-seal.timer` (every 10 min) drives [`scripts/prod-real-seal.sh`](scripts/prod-real-seal.sh), which seals the **current chain epoch** with `block_b = LastEpochBlock` (the epoch's start block — exactly the metagraph both challenges pin their leaf sets against, so D24 participant matching holds by construction). The attempt 409s until both challenges have emitted for that epoch; that is the expected steady state. The gateway prefers chain-scale bundles over the reserved smoke range (`>= 8_000_000`), so once a real seal lands it outranks every interim burn bundle — retire the burn timer (`systemctl disable --now base-burn-seal.timer`) after the first real seal verifies end-to-end. Install: +**Real-epoch sealer (post burn-seal retirement):** `base-real-seal.timer` (every 10 min) drives [`scripts/prod-real-seal.sh`](scripts/prod-real-seal.sh), which walks **current … current−N** chain epochs (`REAL_SEAL_WALK_BACK`, default 16) with `block_b = LastEpochBlock − k×tempo` so a skipped design/prism leaf epoch does not pin `/v1/weights/latest` on a stale real seal forever (burn seals cannot outrank chain-scale bundles). 409 `incomplete_participant_set` on a candidate is expected and the script continues walking. The gateway prefers chain-scale bundles over the reserved smoke range (`>= 8_000_000`) — retire the burn timer (`systemctl disable --now base-burn-seal.timer`) after the first real seal verifies end-to-end. Install: ```bash install -m 0755 deploy/scripts/prod-real-seal.sh /opt/base/deploy/scripts/prod-real-seal.sh diff --git a/deploy/scripts/prod-real-seal.sh b/deploy/scripts/prod-real-seal.sh index 626c70414..9a44f8261 100755 --- a/deploy/scripts/prod-real-seal.sh +++ b/deploy/scripts/prod-real-seal.sh @@ -1,18 +1,15 @@ #!/usr/bin/env bash -# Prod real-epoch sealer: seal the CURRENT chain epoch on the master gateway -# as soon as both >0-bps challenges (design + prism) have posted their leaf -# sets for it. This replaces the interim block-scale burn-seal as the source -# of served weights once real scores flow. +# Prod real-epoch sealer: seal the newest chain epoch on the master gateway +# that has a complete D24 participant set from every >0-bps challenge. # -# Why a blind retry loop: the seal endpoint is fail-safe — it 409s with -# `IncompleteParticipantSet` until both same-epoch sets exist, and re-sealing -# an already-sealed epoch is a no-op conflict. So we simply attempt the -# current epoch every 10 min and log the outcome. +# Why a walk-back: design historically emitted only in a tight late-tempo +# window and could skip alternate epochs (end-of-epoch relabel race). Waiting +# solely on *current* epoch then 409s forever while `/v1/weights/latest` stays +# pinned on an older real seal (burn seals cannot outrank it). Trying current, +# then current-1 … recovers the newest sealable epoch. # -# block_b pins the bundle metagraph. Both challenges pin their expected set -# at the epoch's start block (`last_epoch_block`), so block_b = the current -# LastEpochBlock gives an exact D24 participant match by construction — -# metagraph churn *inside* the epoch can no longer break the seal. +# block_b pins the bundle metagraph to that epoch's start block +# (LastEpochBlock − k×tempo) so D24 participant matching holds. # # Chain reads use plain HTTPS JSON-RPC state_getStorage with baked Substrate # storage keys (twox128("SubtensorModule") ++ twox128(item) ++ netuid LE; @@ -24,12 +21,16 @@ GATEWAY="${BASE_GATEWAY_ENDPOINT:-http://127.0.0.1:8080}" NETUID="${BASE_NETUID:-100}" LOG="${REAL_SEAL_LOG:-/var/log/base-real-seal.log}" LOCK="${REAL_SEAL_LOCK:-/run/base-real-seal.lock}" +# How many prior epochs to try when current is incomplete (≈12h at tempo 360). +WALK_BACK="${REAL_SEAL_WALK_BACK:-16}" # Ordered failover; first reachable endpoint wins per call. CHAIN_ENDPOINTS="${BASE_CHAIN_ENDPOINTS:-https://bittensor-finney.api.onfinality.io/public-ws,https://entrypoint-finney.opentensor.ai:443}" # twox128("SubtensorModule") ++ twox128(item) prefixes (verified on finney). K_SUBNET_EPOCH_INDEX="658faa385070e074c85bf6b568cf05554f101d7a30ae31c7ab3099206c5ae12b" K_LAST_EPOCH_BLOCK="658faa385070e074c85bf6b568cf055590010c37124c14146041452f9ffba0df" +# twox128(SubtensorModule) ++ twox128(Tempo) +K_TEMPO="658faa385070e074c85bf6b568cf05557641384bb339f3758acddfd7053d3317" # Substrate Identity hasher on u16 netuid = little-endian bytes (not printf %04x). netuid_le_hex() { @@ -54,6 +55,53 @@ rpc_storage() { return 1 } +# Tempo is Option on chain (0x01 + LE u16) or bare u16 depending on codec; +# accept both shapes. +rpc_tempo() { + local key="$1" ep out raw + local -a eps + IFS=',' read -r -a eps <<<"${CHAIN_ENDPOINTS}" + for ep in "${eps[@]}"; do + out="$(curl -fsS -m 15 -H 'content-type: application/json' \ + -d "{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"state_getStorage\",\"params\":[\"${key}\"]}" \ + "${ep}" 2>/dev/null)" || continue + raw="$(printf '%s' "${out}" | jq -r '.result // empty')" + if [[ -n "${raw}" && "${raw}" != "null" ]]; then + python3 -c ' +import sys +b=bytes.fromhex(sys.argv[1][2:]) +# Plain u16 (ValueQuery) or Option (0x01 + LE). +if len(b)==3 and b[0]==1: + print(int.from_bytes(b[1:], "little")) +elif len(b)>=2: + print(int.from_bytes(b[:2], "little")) +else: + sys.exit(1) +' "${raw}" && return 0 + fi + done + return 1 +} + +attempt_seal() { + local epoch="$1" block_b="$2" + local resp rc=0 + resp="$(curl -sS -m 60 -X POST -H 'content-type: application/json' \ + -w '\n%{http_code}' \ + ${auth_args[@]+"${auth_args[@]}"} \ + -d "{\"epoch\":${epoch},\"netuid\":${NETUID},\"block_b\":${block_b}}" \ + "${GATEWAY}/v1/admin/seal" 2>&1)" || rc=$? + local http body + http="$(printf '%s' "${resp}" | tail -1)" + body="$(printf '%s' "${resp}" | sed '$d')" + if [[ ${rc} -eq 0 && "${http}" == "200" ]]; then + echo "$(date -Is) seal ok epoch=${epoch} block_b=${block_b}: ${body}" + return 0 + fi + echo "$(date -Is) seal pending/failed epoch=${epoch} block_b=${block_b} http=${http:-?} rc=${rc}: ${body}" + return 1 +} + exec 9>"${LOCK}" if ! flock -n 9; then echo "$(date -Is) skip: another run holds ${LOCK}" >>"${LOG}" @@ -70,6 +118,12 @@ fi echo "$(date -Is) chain read failed (last_epoch_block) key=0x${K_LAST_EPOCH_BLOCK}${netuid_hex}" exit 1 } + # Tempo storage key: twox128(SubtensorModule)++twox128(Tempo)++netuid LE. + # Fallback 360 (finney default) if the read fails. + tempo="$(rpc_tempo "0x${K_TEMPO}${netuid_hex}" 2>/dev/null || true)" + if [[ -z "${tempo}" || "${tempo}" -le 0 ]]; then + tempo=360 + fi auth_args=() if [[ -n "${BASE_GATEWAY_ADMIN_TOKEN:-}" ]]; then auth_args=(-H "Authorization: Bearer ${BASE_GATEWAY_ADMIN_TOKEN}") @@ -78,15 +132,27 @@ fi elif [[ -f "${BASE_HOME}/deploy/secrets/gateway_admin_token" ]]; then auth_args=(-H "Authorization: Bearer $(tr -d '[:space:]' <"${BASE_HOME}/deploy/secrets/gateway_admin_token")") fi - resp="$(curl -fsS -m 60 -X POST -H 'content-type: application/json' \ - ${auth_args[@]+"${auth_args[@]}"} \ - -d "{\"epoch\":${epoch},\"netuid\":${NETUID},\"block_b\":${leb}}" \ - "${GATEWAY}/v1/admin/seal" 2>&1)" && rc=0 || rc=$? - if [[ ${rc} -eq 0 ]]; then - echo "$(date -Is) seal ok epoch=${epoch} block_b=${leb}: ${resp}" - else - # 409 (sets incomplete / already sealed) is the expected steady state - # while waiting on a challenge emission; anything else needs a look. - echo "$(date -Is) seal pending/failed rc=${rc} epoch=${epoch} block_b=${leb}: ${resp}" + + echo "$(date -Is) seal walk start chain_epoch=${epoch} last_epoch_block=${leb} tempo=${tempo} walk_back=${WALK_BACK}" + + sealed=0 + for ((k=0; k<=WALK_BACK; k++)); do + try_epoch=$((epoch - k)) + if (( try_epoch <= 0 )); then + break + fi + try_block=$((leb - k * tempo)) + if (( try_block < 0 )); then + break + fi + if attempt_seal "${try_epoch}" "${try_block}"; then + sealed=1 + break + fi + done + + if [[ "${sealed}" -ne 1 ]]; then + echo "$(date -Is) seal walk exhausted: no complete D24 set in last ${WALK_BACK} epochs (latest remains stale until design+prism emit)" + exit 1 fi } >>"${LOG}" 2>&1 From 63a216ff63b6cbead6f7fcafef16a3b49f8d37d1 Mon Sep 17 00:00:00 2001 From: echobt <154886644+echobt@users.noreply.github.com> Date: Mon, 10 Aug 2026 18:50:49 +0000 Subject: [PATCH 2/2] fix(design): move emit plan out of orchestrator for loc-cap Keep pinned-epoch emit + catch-up scheduling while landing under the design-challenge 1500 LOC gate so CI can merge the seal sync fix. --- crates/design-challenge-task/src/emit.rs | 89 ++++++++++ crates/design-challenge-task/src/lib.rs | 23 ++- crates/design-challenge/src/lib.rs | 15 +- crates/design-challenge/src/orchestrator.rs | 184 +++----------------- 4 files changed, 141 insertions(+), 170 deletions(-) create mode 100644 crates/design-challenge-task/src/emit.rs diff --git a/crates/design-challenge-task/src/emit.rs b/crates/design-challenge-task/src/emit.rs new file mode 100644 index 000000000..948e440a6 --- /dev/null +++ b/crates/design-challenge-task/src/emit.rs @@ -0,0 +1,89 @@ +//! Design leaf-emit scheduling (late-tempo filler + catch-up). + +/// How many blocks before epoch end the NotAttempted filler may run. +/// +/// Wider than the historical 48-block window so `base-real-seal` (10 min) still +/// has time to seal after design emits, while leaving most of the epoch for +/// `award_round` to land Score leaves first (first-write-wins). +pub const DESIGN_EMIT_LATE_BLOCKS: u64 = 96; + +/// Planned design leaf emission for one emitter tick. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct DesignEmitPlan { + /// Epoch label for the leaf set. + pub epoch: u64, + /// Metagraph pin block (epoch start). + pub pin_block: u64, +} + +/// Decide whether/which epoch the design filler should emit. +/// +/// - Catch up `last_emitted+1` when behind by more than one epoch (repairs the +/// end-of-epoch relabel race that skipped alternate epochs in prod). +/// - Otherwise wait until the last [`DESIGN_EMIT_LATE_BLOCKS`] of the current +/// epoch so admin awards can submit Score leaves first. +#[must_use] +pub fn design_emit_plan( + last_emitted: u64, + current_epoch: u64, + blocks_since_last_step: u64, + tempo: u64, + current_last_epoch_block: u64, +) -> Option { + if current_epoch == 0 { + return None; + } + let tempo = tempo.max(1); + if last_emitted >= current_epoch { + return None; + } + // Sequential catch-up for skipped epochs (award path / boundary race). + if last_emitted + 1 < current_epoch { + let target = last_emitted + 1; + let epochs_back = current_epoch.saturating_sub(target); + let pin_block = current_last_epoch_block.saturating_sub(epochs_back.saturating_mul(tempo)); + return Some(DesignEmitPlan { + epoch: target, + pin_block, + }); + } + // Current epoch: late-tempo filler only. + if blocks_since_last_step.saturating_add(DESIGN_EMIT_LATE_BLOCKS) < tempo { + return None; + } + Some(DesignEmitPlan { + epoch: current_epoch, + pin_block: current_last_epoch_block, + }) +} + +#[cfg(test)] +mod tests { + #![allow(clippy::unwrap_used)] + use super::*; + + #[test] + fn emit_plan_waits_until_late_tempo_for_current_epoch() { + assert!(design_emit_plan(10, 11, 200, 360, 1000).is_none()); + let p = design_emit_plan(10, 11, 360 - DESIGN_EMIT_LATE_BLOCKS, 360, 1000).unwrap(); + assert_eq!( + p, + DesignEmitPlan { + epoch: 11, + pin_block: 1000 + } + ); + } + + #[test] + fn emit_plan_catches_up_skipped_epochs_without_waiting() { + let p = design_emit_plan(24412, 24423, 50, 360, 8_815_687).unwrap(); + assert_eq!(p.epoch, 24413); + assert_eq!(p.pin_block, 8_815_687 - (24423 - 24413) * 360); + } + + #[test] + fn emit_plan_noop_when_already_emitted_current() { + assert!(design_emit_plan(11, 11, 350, 360, 1000).is_none()); + } +} diff --git a/crates/design-challenge-task/src/lib.rs b/crates/design-challenge-task/src/lib.rs index 9df60bb8e..562c21ddd 100644 --- a/crates/design-challenge-task/src/lib.rs +++ b/crates/design-challenge-task/src/lib.rs @@ -11,7 +11,9 @@ #![forbid(unsafe_code)] #![allow(clippy::missing_errors_doc, clippy::doc_markdown)] +mod emit; mod score; +pub use emit::{design_emit_plan, DesignEmitPlan, DESIGN_EMIT_LATE_BLOCKS}; pub use score::{ not_attempted, round_win_delta, score_window, to_leaf, window_start, ScorePlan, WindowScorePlan, }; @@ -77,7 +79,8 @@ pub const fn unscored_epochs_elapsed(start_epoch: u64, current_epoch: u64) -> bo current_epoch.saturating_sub(start_epoch) >= UNSCORED_EPOCH_LIMIT } -fn now_ms() -> u64 { +#[must_use] +pub fn now_ms() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)) @@ -320,6 +323,24 @@ pub const fn rounds_for_day(day_index: u64) -> (u64, u64) { (start, start + ROUNDS_PER_DAY - 1) } +/// Cap harness log payload stored in stage-event detail (JSON). +pub const MAX_LOG_CHARS: usize = 65_536; + +#[must_use] +pub fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()) +} + +#[must_use] +pub fn clip_logs(text: &str) -> String { + if text.len() <= MAX_LOG_CHARS { + return text.to_owned(); + } + format!("...[truncated]\n{}", &text[text.len() - MAX_LOG_CHARS..]) +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/design-challenge/src/lib.rs b/crates/design-challenge/src/lib.rs index e117631f9..a7c319460 100644 --- a/crates/design-challenge/src/lib.rs +++ b/crates/design-challenge/src/lib.rs @@ -27,12 +27,13 @@ pub use challenge_common::{ GatewayClient, GatewayClientConfig, LeafEmitError, }; pub use design_challenge_task::{ - agent_run_timeout_secs, awaiting_admin_unscored_expired, daily_run_quota, + agent_run_timeout_secs, awaiting_admin_unscored_expired, daily_run_quota, design_emit_plan, manual_daily_run_quota, prompts_per_round, reject_awaiting_admin_run, round_id_at, round_secs, round_win_delta, rounds_per_day_effective, scheduled_daily_run_cap, scheduled_runs_per_day, - score_window, unscored_epochs_elapsed, window_start, ScorePlan, WindowScorePlan, CHALLENGE_ID, - CHALLENGE_ID_BYTES, MANUAL_DAILY_RUN_QUOTA, PROMPTS_PER_ROUND, ROUNDS_PER_DAY, ROUND_SECS, - SCORE_MAX, SCORING_VERSION, SCORING_WINDOW_ROUNDS, UNSCORED_EPOCH_LIMIT, + score_window, unscored_epochs_elapsed, window_start, DesignEmitPlan, ScorePlan, + WindowScorePlan, CHALLENGE_ID, CHALLENGE_ID_BYTES, DESIGN_EMIT_LATE_BLOCKS, + MANUAL_DAILY_RUN_QUOTA, PROMPTS_PER_ROUND, ROUNDS_PER_DAY, ROUND_SECS, SCORE_MAX, + SCORING_VERSION, SCORING_WINDOW_ROUNDS, UNSCORED_EPOCH_LIMIT, }; pub use design_http::{ design_router, mark_awaiting, mark_awaiting_admin, record_epoch, AdminAwardHook, AppState, @@ -44,12 +45,8 @@ pub use design_store_pg::DbDesignStore; pub use host_sim::{ force_sim_refusal_reason, host_sim_allowed, is_prod_env, require_host_sim_for_force, }; -pub use orchestrator::{ - design_emit_plan, DesignEmitPlan, ErrorClass, Orchestrator, OrchestratorConfig, - DESIGN_EMIT_LATE_BLOCKS, -}; +pub use orchestrator::{ErrorClass, Orchestrator, OrchestratorConfig}; -/// Crate identity smoke. #[must_use] pub fn crate_name() -> &'static str { "design-challenge" diff --git a/crates/design-challenge/src/orchestrator.rs b/crates/design-challenge/src/orchestrator.rs index 085621362..7d94f1c3c 100644 --- a/crates/design-challenge/src/orchestrator.rs +++ b/crates/design-challenge/src/orchestrator.rs @@ -12,13 +12,14 @@ use challenge_agentic::{ copy_gate, AgenticBackend, AgenticError, AgenticVerdict, ReviewRequest, VerdictKind, }; use challenge_common::{ - emit_signed_leaf_set, expected_set_at_chain, submit_signed_leaf_set, ExpectedSet, - GatewayClient, PinnedBlockHash, + emit_signed_leaf_set, expected_set_at_chain, submit_signed_leaf_set, GatewayClient, + PinnedBlockHash, }; use crypto::KEY_LEN; use design_challenge_task::{ - awaiting_admin_unscored_expired, reject_awaiting_admin_run, round_id_at, round_secs, - UNSCORED_EPOCH_LIMIT, + awaiting_admin_unscored_expired, clip_logs, design_emit_plan, not_attempted, now_ms, now_secs, + reject_awaiting_admin_run, round_id_at, round_secs, score_window, to_leaf, window_start, + WindowScorePlan, MAX_LOG_CHARS, UNSCORED_EPOCH_LIMIT, }; use design_http::{mark_awaiting_admin, AdminAwardHook}; use design_prompts::{prompt_set_digest, select_prompts_for_round}; @@ -35,27 +36,16 @@ use tracing::{info, warn}; use crate::corpus; use crate::screenshot::{capture_full_page_png, png_artifact_tuple}; use crate::CHALLENGE_ID; -use design_challenge_task::{not_attempted, score_window, to_leaf, window_start, WindowScorePlan}; -/// Cap harness log payload stored in stage-event detail (JSON). -const MAX_LOG_CHARS: usize = 65_536; - -/// Classified run failure: retryable infra classes vs terminal miner errors. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ErrorClass { - /// Package install / Docker engine infra (auto-retryable). Install, - /// AST / workdir / store infra (auto-retryable). AstInfra, - /// LLM transport / provider / verdict parse (auto-retryable). LlmInfra, - /// Miner-caused failure: agent crash, sanitize reject, missing output - /// (terminal — no auto-retry). Miner, } impl ErrorClass { - /// DB / gating label. #[must_use] pub const fn as_str(self) -> &'static str { match self { @@ -66,7 +56,6 @@ impl ErrorClass { } } - /// Auto-retry budget applies (initial attempt + up to 3 retries). #[must_use] pub const fn retryable(self) -> bool { !matches!(self, Self::Miner) @@ -121,7 +110,6 @@ fn classify_agentic(e: &AgenticError) -> RunFailure { } } -/// Design domain rules appended to the agentic system prompt. const DESIGN_AGENTIC_RULES: &str = r" Design challenge rules: - Allowed: PyPI deps, external APIs/MCP over egress (network ≠ cheat), Mobbin/Dribbble, image gen, UI libs. @@ -130,26 +118,15 @@ Design challenge rules: - suspicious and cheat → Score(0), not admin-eligible. "; -/// Orchestrator config. #[derive(Debug, Clone)] pub struct OrchestratorConfig { - /// Netuid. pub netuid: u16, - /// Claim poll. pub claim_poll: Duration, - /// Stuck grace secs. pub stuck_grace_secs: u64, - /// LLM proxy URL for sandbox. pub llm_proxy: String, - /// Staging root for agentic workdirs. pub staging_root: PathBuf, - /// Local/e2e only: pause after each published stage so mid-flight is - /// photographable. Zero in production (default). pub stage_delay: Duration, - /// Auto-retry budget for infra-class failures (initial attempt + this - /// many retries). Default 3; `cheat` / `rejected` are always terminal. pub auto_retry_max: u32, - /// Poll interval for the late-tempo D24 filler. pub emit_poll: Duration, } @@ -168,7 +145,6 @@ impl Default for OrchestratorConfig { } } -/// Orchestrator handle. pub struct Orchestrator { cfg: OrchestratorConfig, store: Arc, @@ -178,9 +154,7 @@ pub struct Orchestrator { chain: Arc, sk: [u8; KEY_LEN], gating: Option>, - /// Shared chain-epoch cache (HTTP `AppState.epoch` + sweeper clock). epoch_cache: Option>, - /// Last epoch successfully covered by [`Self::emit_leaves`]. emitted_epoch: AtomicU64, } @@ -193,7 +167,6 @@ impl std::fmt::Debug for Orchestrator { } impl Orchestrator { - /// Construct. #[must_use] pub fn new( cfg: OrchestratorConfig, @@ -218,7 +191,6 @@ impl Orchestrator { } } - /// Attach the submission gating store (terminal states + retry attempts). #[must_use] pub fn with_gating(mut self, gating: Arc) -> Self { self.gating = Some(gating); @@ -241,14 +213,12 @@ impl Orchestrator { epoch } - /// Local/e2e: hold after publishing a stage so polls can photograph it. async fn pause_stage(&self) { if !self.cfg.stage_delay.is_zero() { sleep(self.cfg.stage_delay).await; } } - /// Worker loop. pub async fn run_worker(self: Arc) { loop { match self.cycle_once().await { @@ -262,7 +232,6 @@ impl Orchestrator { } } - /// Round closer + leaf emitter loop. pub async fn run_round_loop(self: Arc) { let mut last_closed = 0u64; loop { @@ -285,8 +254,6 @@ impl Orchestrator { } } - /// D24 filler + catch-up: covers epochs with no admin award, and repairs - /// gaps left by the end-of-epoch boundary race (see [`design_emit_plan`]). pub async fn run_emitter(self: Arc) where C: Sync, @@ -299,31 +266,21 @@ impl Orchestrator { } } - /// One emitter tick. `Ok(true)` when a leaf set was submitted this tick. - /// - /// # Errors - /// Chain / sign / submit failures (retried next tick). pub async fn emitter_tick(&self) -> Result where C: Sync, { let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) .map_err(|e| format!("schedule: {e}"))?; - let epoch = state.subnet_epoch_index; - let tempo = u64::from(state.tempo.max(1)); - let last = self.emitted_epoch.load(Ordering::Relaxed); let Some(plan) = design_emit_plan( - last, - epoch, + self.emitted_epoch.load(Ordering::Relaxed), + state.subnet_epoch_index, state.blocks_since_last_step, - tempo, + u64::from(state.tempo.max(1)), state.last_epoch_block, ) else { return Ok(false); }; - // Pin epoch + block from this tick's schedule snapshot — do **not** - // re-read chain inside emit (end-of-epoch flip used to relabel the set - // as E+1 and permanently skip E, starving real-seal with D24 409s). self.emit_leaves_at(plan.epoch, plan.pin_block).await?; Ok(true) } @@ -361,7 +318,6 @@ impl Orchestrator { Ok(()) } - /// Stuck sweeper + unscored-timeout sweep. pub async fn run_sweeper(self: Arc) { loop { sleep(Duration::from_secs(60)).await; @@ -392,7 +348,6 @@ impl Orchestrator { } } - /// Award winners (admin API) or timeout close: score + emit leaves. pub async fn award_round(&self, rid: u64) -> Result<(), String> { let _ = self.ensure_round(rid).await; let _ = self.store.set_round_status(rid, "scoring").await; @@ -556,15 +511,14 @@ impl Orchestrator { } } - self.emit_leaves().await?; + let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) + .map_err(|e| format!("schedule: {e}"))?; + self.emit_leaves_at(state.subnet_epoch_index, state.last_epoch_block) + .await?; let _ = self.store.set_round_status(rid, "emitted").await; Ok(()) } - /// One claim→execute cycle; `Ok(true)` when one run was worked. - /// - /// # Errors - /// Claim fault only; per-run failures are retried or finalized inline. pub async fn cycle_once(&self) -> Result { let Some(run) = self .store @@ -597,8 +551,6 @@ impl Orchestrator { Ok(true) } - /// Auto-retry retryable infra classes up to `auto_retry_max`; otherwise - /// finalize `failed` and block the hotkey in the gating store. async fn handle_run_failure(&self, run: &design_store::RunState, f: RunFailure) { let hotkey = self .store @@ -1082,7 +1034,6 @@ impl Orchestrator { Ok(()) } - /// Close round: if winners already awarded/emitted, no-op; else award (timeout zeros). async fn close_round(&self, rid: u64) -> Result<(), String> { if let Ok(Some(r)) = self.store.get_round(rid).await { if r.status == "emitted" { @@ -1093,17 +1044,6 @@ impl Orchestrator { self.award_round(rid).await } - async fn emit_leaves(&self) -> Result<(), String> { - let state = chain::gather_schedule_state(self.chain.as_ref(), self.cfg.netuid) - .map_err(|e| format!("schedule: {e}"))?; - self.emit_leaves_at(state.subnet_epoch_index, state.last_epoch_block) - .await - } - - /// Submit a D24-complete design leaf set for a pinned `(epoch, pin_block)`. - /// - /// `pin_block` must be that epoch's start block (`LastEpochBlock` while the - /// epoch is current, or `current_last_epoch_block - k*tempo` when catching up). async fn emit_leaves_at(&self, epoch: u64, pin_block: u64) -> Result<(), String> { if epoch == 0 { return Err("refuse emit for epoch 0".into()); @@ -1112,27 +1052,29 @@ impl Orchestrator { .chain .block_hash(pin_block) .map_err(|e| format!("block_hash@{pin_block}: {e}"))?; - let expected: ExpectedSet = expected_set_at_chain( + let expected = expected_set_at_chain( &trustroot::ParticipantPolicy::AllMetagraphHotkeys, PinnedBlockHash::new(block_hash), self.chain.as_ref(), ) .map_err(|e| format!("expected set: {e}"))?; - let stored = self + let by_miner: BTreeMap<_, _> = self .store .scores_for_epoch(self.cfg.netuid, epoch) .await - .map_err(|e| e.to_string())?; - let by_miner: BTreeMap = stored.into_iter().collect(); + .map_err(|e| e.to_string())? + .into_iter() + .collect(); let mut scores = BTreeMap::new(); - let mut expected_set: BTreeSet<[u8; KEY_LEN]> = BTreeSet::new(); + let mut expected_set = BTreeSet::new(); for p in &expected.participants { expected_set.insert(p.hotkey); - let leaf = match by_miner.get(&hex::encode(p.hotkey)) { - Some(fs) => to_leaf(fs), - None => to_leaf(¬_attempted()), - }; - scores.insert(p.hotkey, leaf); + scores.insert( + p.hotkey, + by_miner + .get(&hex::encode(p.hotkey)) + .map_or_else(|| to_leaf(¬_attempted()), to_leaf), + ); } let signed = emit_signed_leaf_set( &self.sk, @@ -1156,87 +1098,9 @@ impl Orchestrator { } } -/// How many blocks before epoch end the NotAttempted filler may run. -/// -/// Wider than the historical 48-block window so `base-real-seal` (10 min) still -/// has time to seal after design emits, while leaving most of the epoch for -/// `award_round` to land Score leaves first (first-write-wins). -pub const DESIGN_EMIT_LATE_BLOCKS: u64 = 96; - -/// Planned design leaf emission for one emitter tick. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct DesignEmitPlan { - /// Epoch label for the leaf set. - pub epoch: u64, - /// Metagraph pin block (epoch start). - pub pin_block: u64, -} - -/// Decide whether/which epoch the design filler should emit. -/// -/// - Catch up `last_emitted+1` when behind by more than one epoch (repairs the -/// end-of-epoch relabel race that skipped alternate epochs in prod). -/// - Otherwise wait until the last [`DESIGN_EMIT_LATE_BLOCKS`] of the current -/// epoch so admin awards can submit Score leaves first. -#[must_use] -pub fn design_emit_plan( - last_emitted: u64, - current_epoch: u64, - blocks_since_last_step: u64, - tempo: u64, - current_last_epoch_block: u64, -) -> Option { - if current_epoch == 0 { - return None; - } - let tempo = tempo.max(1); - if last_emitted >= current_epoch { - return None; - } - // Sequential catch-up for skipped epochs (award path / boundary race). - if last_emitted + 1 < current_epoch { - let target = last_emitted + 1; - let epochs_back = current_epoch.saturating_sub(target); - let pin_block = current_last_epoch_block.saturating_sub(epochs_back.saturating_mul(tempo)); - return Some(DesignEmitPlan { - epoch: target, - pin_block, - }); - } - // Current epoch: late-tempo filler only. - if blocks_since_last_step.saturating_add(DESIGN_EMIT_LATE_BLOCKS) < tempo { - return None; - } - Some(DesignEmitPlan { - epoch: current_epoch, - pin_block: current_last_epoch_block, - }) -} - #[async_trait] impl AdminAwardHook for Orchestrator { async fn on_winners(&self, round_id: u64, _harness_ids: &[String]) -> Result<(), String> { self.award_round(round_id).await } } - -fn now_secs() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.as_secs()) -} - -fn now_ms() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX)) -} - -fn clip_logs(text: &str) -> String { - if text.len() <= MAX_LOG_CHARS { - return text.to_owned(); - } - let mut out = text[text.len() - MAX_LOG_CHARS..].to_owned(); - out.insert_str(0, "...[truncated]\n"); - out -}