feat(consensus): P2P consensus system with SDK trait refactor and WASM weight integration

bins/validator-node/src/challenge_storage.rs

@@ -54,4 +54,12 @@ impl StorageBackend for ChallengeStorageBackend {
             .block_on(self.storage.delete(&storage_key))
             .map_err(|e| StorageHostError::StorageError(e.to_string()))
     }
+
+    fn get_cross(
+        &self,
+        challenge_id: &str,
+        key: &[u8],
+    ) -> Result<Option<Vec<u8>>, StorageHostError> {
+        self.get(challenge_id, key)
+    }
 }

bins/validator-node/src/main.rs

@@ -267,7 +267,7 @@ async fn main() -> Result<()> {
         .with_listen_addr(&args.listen_addr)
         .with_bootstrap_peers(args.bootstrap.clone())
         .with_netuid(args.netuid)
-        .with_min_stake(1_000_000_000_000); // 1000 TAO
+        .with_min_stake(10_000_000_000_000); // 10000 TAO
 
     // Initialize validator set (ourselves first)
     let validator_set = Arc::new(ValidatorSet::new(keypair.clone(), p2p_config.min_stake));
@@ -478,6 +478,7 @@ async fn main() -> Result<()> {
                     &consensus,
                     &validator_set,
                     &state_manager,
+                    &wasm_executor,
                 ).await;
             }
 
@@ -506,6 +507,8 @@ async fn main() -> Result<()> {
                     &state_manager,
                     netuid,
                     version_key,
+                    &wasm_executor,
+                    &keypair,
                 ).await;
             }
 
@@ -689,6 +692,7 @@ async fn handle_network_event(
     consensus: &Arc<RwLock<ConsensusEngine>>,
     validator_set: &Arc<ValidatorSet>,
     state_manager: &Arc<StateManager>,
+    wasm_executor_ref: &Option<Arc<WasmChallengeExecutor>>,
 ) {
     match event {
         NetworkEvent::Message { source, message } => match message {
@@ -969,13 +973,27 @@ async fn handle_network_event(
                 );
             }
             P2PMessage::ChallengeUpdate(update) => {
-                info!(
-                    challenge_id = %update.challenge_id,
-                    updater = %update.updater.to_hex(),
-                    update_type = %update.update_type,
-                    data_bytes = update.data.len(),
-                    "Received challenge update"
-                );
+                let updater_ss58 = update.updater.to_hex();
+                if updater_ss58 == platform_p2p_consensus::SUDO_HOTKEY
+                    || update.updater.0 == platform_core::SUDO_KEY_BYTES
+                {
+                    info!(
+                        challenge_id = %update.challenge_id,
+                        updater = %updater_ss58,
+                        update_type = %update.update_type,
+                        data_bytes = update.data.len(),
+                        "Received authorized challenge update from sudo key"
+                    );
+                    if let Some(ref executor) = wasm_executor_ref {
+                        executor.invalidate_cache(&update.challenge_id.to_string());
+                    }
+                } else {
+                    warn!(
+                        challenge_id = %update.challenge_id,
+                        updater = %updater_ss58,
+                        "Rejected challenge update from non-sudo key"
+                    );
+                }
             }
             P2PMessage::StorageProposal(proposal) => {
                 debug!(
@@ -1059,6 +1077,8 @@ async fn handle_block_event(
     state_manager: &Arc<StateManager>,
     netuid: u16,
     version_key: u64,
+    wasm_executor: &Option<Arc<WasmChallengeExecutor>>,
+    keypair: &Keypair,
 ) {
     match event {
         BlockSyncEvent::NewBlock { block_number, .. } => {
@@ -1089,6 +1109,45 @@ async fn handle_block_event(
                 epoch, block
             );
 
+            // Collect WASM-computed weights from challenges before finalizing
+            if let Some(ref executor) = wasm_executor {
+                let challenges: Vec<String> = state_manager
+                    .apply(|state| state.challenges.keys().map(|k| k.to_string()).collect());
+                let local_hotkey = keypair.hotkey();
+                for cid in &challenges {
+                    match executor.execute_get_weights(cid) {
+                        Ok(weights) if !weights.is_empty() => {
+                            state_manager.apply(|state| {
+                                if let Err(e) = state.submit_weight_vote(
+                                    local_hotkey.clone(),
+                                    netuid,
+                                    weights.clone(),
+                                ) {
+                                    warn!(
+                                        challenge_id = %cid,
+                                        error = %e,
+                                        "Failed to submit WASM-computed weights"
+                                    );
+                                }
+                            });
+                            info!(
+                                challenge_id = %cid,
+                                weight_count = weights.len(),
+                                "Integrated WASM-computed weights"
+                            );
+                        }
+                        Ok(_) => {}
+                        Err(e) => {
+                            debug!(
+                                challenge_id = %cid,
+                                error = %e,
+                                "WASM get_weights not available for challenge"
+                            );
+                        }
+                    }
+                }
+            }
+
             // Get weights from decentralized state
             if let (Some(st), Some(sig)) = (subtensor.as_ref(), signer.as_ref()) {
                 let final_weights = state_manager.apply(|state| state.finalize_weights());

bins/validator-node/src/wasm_executor.rs

@@ -793,6 +793,116 @@ impl WasmChallengeExecutor {
         Ok((result_data, metrics))
     }
 
+    pub fn execute_get_weights(&self, module_path: &str) -> Result<Vec<(u16, u16)>> {
+        let start = Instant::now();
+
+        let module = self
+            .load_module(module_path)
+            .context("Failed to load WASM module")?;
+
+        let network_host_fns = Arc::new(NetworkHostFunctions::all());
+
+        let instance_config = InstanceConfig {
+            challenge_id: module_path.to_string(),
+            validator_id: "validator".to_string(),
+            storage_host_config: StorageHostConfig {
+                allow_direct_writes: true,
+                require_consensus: false,
+                ..self.config.storage_host_config.clone()
+            },
+            storage_backend: Arc::clone(&self.config.storage_backend),
+            consensus_policy: ConsensusPolicy::read_only(),
+            ..Default::default()
+        };
+
+        let mut instance = self
+            .runtime
+            .instantiate(&module, instance_config, Some(network_host_fns))
+            .map_err(|e| anyhow::anyhow!("WASM instantiation failed: {}", e))?;
+
+        let result = instance
+            .call_return_i64("get_weights")
+            .map_err(|e| anyhow::anyhow!("WASM get_weights call failed: {}", e))?;
+
+        let out_len = (result >> 32) as i32;
+        let out_ptr = (result & 0xFFFF_FFFF) as i32;
+
+        let result_data = if out_ptr > 0 && out_len > 0 {
+            instance
+                .read_memory(out_ptr as usize, out_len as usize)
+                .map_err(|e| {
+                    anyhow::anyhow!("failed to read WASM memory for get_weights output: {}", e)
+                })?
+        } else {
+            return Ok(Vec::new());
+        };
+
+        let weights: Vec<(u16, u16)> = bincode::DefaultOptions::new()
+            .with_fixint_encoding()
+            .allow_trailing_bytes()
+            .with_limit(MAX_ROUTE_OUTPUT_SIZE)
+            .deserialize(&result_data)
+            .context("Failed to deserialize get_weights output")?;
+
+        info!(
+            module = module_path,
+            weight_count = weights.len(),
+            execution_time_ms = start.elapsed().as_millis() as u64,
+            "WASM get_weights completed"
+        );
+
+        Ok(weights)
+    }
+
+    #[allow(dead_code)]
+    pub fn execute_validate_storage_write(
+        &self,
+        module_path: &str,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<bool> {
+        let module = self
+            .load_module(module_path)
+            .context("Failed to load WASM module")?;
+
+        let network_host_fns = Arc::new(NetworkHostFunctions::all());
+
+        let instance_config = InstanceConfig {
+            challenge_id: module_path.to_string(),
+            validator_id: "validator".to_string(),
+            storage_host_config: StorageHostConfig::default(),
+            storage_backend: Arc::clone(&self.config.storage_backend),
+            ..Default::default()
+        };
+
+        let mut instance = self
+            .runtime
+            .instantiate(&module, instance_config, Some(network_host_fns))
+            .map_err(|e| anyhow::anyhow!("WASM instantiation failed: {}", e))?;
+
+        let key_ptr = self.allocate_input(&mut instance, key)?;
+        instance
+            .write_memory(key_ptr as usize, key)
+            .map_err(|e| anyhow::anyhow!("Failed to write key to WASM memory: {}", e))?;
+
+        let val_ptr = self.allocate_input(&mut instance, value)?;
+        instance
+            .write_memory(val_ptr as usize, value)
+            .map_err(|e| anyhow::anyhow!("Failed to write value to WASM memory: {}", e))?;
+
+        let result = instance
+            .call_i32_i32_i32_i32_return_i32(
+                "validate_storage_write",
+                key_ptr,
+                key.len() as i32,
+                val_ptr,
+                value.len() as i32,
+            )
+            .map_err(|e| anyhow::anyhow!("WASM validate_storage_write call failed: {}", e))?;
+
+        Ok(result == 1)
+    }
+
     fn load_module(&self, module_path: &str) -> Result<Arc<WasmModule>> {
         {
             let cache = self.module_cache.read();

crates/challenge-sdk-wasm/src/host_functions.rs

@@ -16,6 +16,13 @@ extern "C" {
 extern "C" {
     fn storage_get(key_ptr: i32, key_len: i32, value_ptr: i32) -> i32;
     fn storage_set(key_ptr: i32, key_len: i32, value_ptr: i32, value_len: i32) -> i32;
+    fn storage_get_cross(
+        cid_ptr: i32,
+        cid_len: i32,
+        key_ptr: i32,
+        key_len: i32,
+        value_ptr: i32,
+    ) -> i32;
 }
 
 #[link(wasm_import_module = "platform_terminal")]
@@ -110,6 +117,24 @@ pub fn host_storage_set(key: &[u8], value: &[u8]) -> Result<(), i32> {
     Ok(())
 }
 
+pub fn host_storage_get_cross(challenge_id: &[u8], key: &[u8]) -> Result<Vec<u8>, i32> {
+    let mut value_buf = vec![0u8; RESPONSE_BUF_MEDIUM];
+    let status = unsafe {
+        storage_get_cross(
+            challenge_id.as_ptr() as i32,
+            challenge_id.len() as i32,
+            key.as_ptr() as i32,
+            key.len() as i32,
+            value_buf.as_mut_ptr() as i32,
+        )
+    };
+    if status < 0 {
+        return Err(status);
+    }
+    value_buf.truncate(status as usize);
+    Ok(value_buf)
+}
+
 pub fn host_terminal_exec(request: &[u8]) -> Result<Vec<u8>, i32> {
     let mut result_buf = vec![0u8; RESPONSE_BUF_LARGE];
     let status = unsafe {
@@ -256,6 +281,7 @@ extern "C" {
     fn consensus_get_state_hash(buf_ptr: i32) -> i32;
     fn consensus_get_submission_count() -> i32;
     fn consensus_get_block_height() -> i64;
+    fn consensus_get_subnet_challenges(buf_ptr: i32, buf_len: i32) -> i32;
 }
 
 pub fn host_consensus_get_epoch() -> i64 {
@@ -306,3 +332,14 @@ pub fn host_consensus_get_submission_count() -> i32 {
 pub fn host_consensus_get_block_height() -> i64 {
     unsafe { consensus_get_block_height() }
 }
+
+pub fn host_consensus_get_subnet_challenges() -> Result<Vec<u8>, i32> {
+    let mut buf = vec![0u8; RESPONSE_BUF_MEDIUM];
+    let status =
+        unsafe { consensus_get_subnet_challenges(buf.as_mut_ptr() as i32, buf.len() as i32) };
+    if status < 0 {
+        return Err(status);
+    }
+    buf.truncate(status as usize);
+    Ok(buf)
+}

crates/challenge-sdk-wasm/src/lib.rs

@@ -13,7 +13,7 @@ pub use types::{
 };
 pub use types::{ContainerRunRequest, ContainerRunResponse};
 pub use types::{EvaluationInput, EvaluationOutput};
-pub use types::{WasmRouteDefinition, WasmRouteRequest, WasmRouteResponse};
+pub use types::{WasmRouteDefinition, WasmRouteRequest, WasmRouteResponse, WeightEntry};
 
 pub trait Challenge {
     fn name(&self) -> &'static str;
@@ -49,6 +49,19 @@ pub trait Challenge {
     fn handle_route(&self, _request: &[u8]) -> alloc::vec::Vec<u8> {
         alloc::vec::Vec::new()
     }
+
+    /// Return serialized epoch weight entries (`Vec<WeightEntry>`) that the
+    /// validator should set on-chain. The default implementation returns an
+    /// empty vector (no weights).
+    fn get_weights(&self) -> alloc::vec::Vec<u8> {
+        alloc::vec::Vec::new()
+    }
+
+    /// Validate whether a storage write with the given `key` and `value` is
+    /// permitted. The default implementation allows all writes.
+    fn validate_storage_write(&self, _key: &[u8], _value: &[u8]) -> bool {
+        true
+    }
 }
 
 /// Pack a pointer and length into a single i64 value.
@@ -264,5 +277,39 @@ macro_rules! register_challenge {
             }
             $crate::pack_ptr_len(ptr as i32, output.len() as i32)
         }
+
+        #[no_mangle]
+        pub extern "C" fn get_weights() -> i64 {
+            let output = <$ty as $crate::Challenge>::get_weights(&_CHALLENGE);
+            if output.is_empty() {
+                return $crate::pack_ptr_len(0, 0);
+            }
+            let ptr = $crate::alloc_impl::sdk_alloc(output.len());
+            if ptr.is_null() {
+                return $crate::pack_ptr_len(0, 0);
+            }
+            unsafe {
+                core::ptr::copy_nonoverlapping(output.as_ptr(), ptr, output.len());
+            }
+            $crate::pack_ptr_len(ptr as i32, output.len() as i32)
+        }
+
+        #[no_mangle]
+        pub extern "C" fn validate_storage_write(
+            key_ptr: i32,
+            key_len: i32,
+            val_ptr: i32,
+            val_len: i32,
+        ) -> i32 {
+            let key =
+                unsafe { core::slice::from_raw_parts(key_ptr as *const u8, key_len as usize) };
+            let value =
+                unsafe { core::slice::from_raw_parts(val_ptr as *const u8, val_len as usize) };
+            if <$ty as $crate::Challenge>::validate_storage_write(&_CHALLENGE, key, value) {
+                1
+            } else {
+                0
+            }
+        }
     };
 }

crates/challenge-sdk-wasm/src/types.rs

@@ -189,3 +189,13 @@ pub struct WasmRouteResponse {
     /// Raw response body bytes.
     pub body: Vec<u8>,
 }
+
+/// A single weight entry mapping a UID to a weight value.
+///
+/// Returned by [`Challenge::get_weights`] as a serialized `Vec<WeightEntry>`.
+/// Both fields use `u16` to match the on-chain weight vector format.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct WeightEntry {
+    pub uid: u16,
+    pub weight: u16,
+}