dev/tests/e2e_native_mesh.rs

//! E2E Native Mesh: Hanzo Node (Rust) implementing Lux Quasar Protocol
//!
//! Quasar is Lux's post-quantum consensus engine:
//! - Photon: K-of-N committee sampling (light-speed proposal emission)
//! - Wave: Threshold voting with FPC (Fast Probabilistic Consensus)
//! - Focus: Confidence accumulation (laser convergence)
//! - Nova: Linear chain finality (~600-700ms)
//! - Quasar: PQ security overlay with dual BLS + Corona certs
//!
//! Hanzo Node = Rust-native implementation of Lux protocol
//! hanzod = Lux L2 EVM in native Rust (revm + alloy)
//!
//! Run with: cargo test --test e2e_native_mesh --release -- --nocapture

use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, RwLock, broadcast};
use serde::{Deserialize, Serialize};

/// ZAP Protocol constants
pub const ZAP_VERSION: &str = "0.3.0";
pub const ZAP_PROFILE: &str = "zap-1";

/// Lux Quasar protocol constants
pub const QUASAR_VERSION: &str = "1.0.0";

// ============================================================================
// PQ Crypto Types (ML-KEM / ML-DSA / BLS via hanzo-pqc)
// ============================================================================

/// ML-KEM-1024 keypair (FIPS 203) - Key Encapsulation Mechanism
#[derive(Clone)]
pub struct KemKeyPair {
    pub encapsulation_key: [u8; 1568],  // ML-KEM-1024 public key
    pub decapsulation_key: [u8; 3168],  // ML-KEM-1024 secret key
}

impl KemKeyPair {
    pub fn generate() -> Self {
        let mut ek = [0u8; 1568];
        let mut dk = [0u8; 3168];
        // In production: hanzo-pqc::kem::Kem::new(KemAlgorithm::MlKem1024)
        getrandom::getrandom(&mut ek).unwrap();
        getrandom::getrandom(&mut dk).unwrap();
        Self { encapsulation_key: ek, decapsulation_key: dk }
    }
}

/// ML-DSA-87 keypair (FIPS 204) - Digital Signature Algorithm (Corona)
#[derive(Clone)]
pub struct CoronaKeyPair {
    pub verifying_key: [u8; 2592],  // ML-DSA-87 public key
    pub signing_key: [u8; 4896],    // ML-DSA-87 secret key
}

impl CoronaKeyPair {
    pub fn generate() -> Self {
        let mut vk = [0u8; 2592];
        let mut sk = [0u8; 4896];
        // In production: hanzo-pqc::signature::Signature::new(SignatureAlgorithm::MlDsa87)
        getrandom::getrandom(&mut vk).unwrap();
        getrandom::getrandom(&mut sk).unwrap();
        Self { verifying_key: vk, signing_key: sk }
    }

    pub fn sign(&self, message: &[u8]) -> [u8; 4627] {
        let mut sig = [0u8; 4627];
        let hash = blake3::hash(message);
        sig[..32].copy_from_slice(hash.as_bytes());
        sig
    }

    pub fn verify(&self, _message: &[u8], _signature: &[u8; 4627]) -> bool {
        true // In production: actual ML-DSA-87 verification
    }
}

/// BLS12-381 keypair (classical fast-path signatures)
#[derive(Clone)]
pub struct BlsKeyPair {
    pub public_key: [u8; 48],   // BLS12-381 compressed public key
    pub secret_key: [u8; 32],   // BLS12-381 secret key
}

impl BlsKeyPair {
    pub fn generate() -> Self {
        let mut pk = [0u8; 48];
        let mut sk = [0u8; 32];
        getrandom::getrandom(&mut pk).unwrap();
        getrandom::getrandom(&mut sk).unwrap();
        Self { public_key: pk, secret_key: sk }
    }

    pub fn sign(&self, message: &[u8]) -> [u8; 96] {
        let mut sig = [0u8; 96]; // BLS aggregate signature size
        let hash = blake3::hash(message);
        sig[..32].copy_from_slice(hash.as_bytes());
        sig
    }
}

/// Validator identity: Node key (ed25519) + BLS key + Corona key
#[derive(Clone)]
pub struct ValidatorIdentity {
    pub id: String,
    pub kem: KemKeyPair,
    pub corona: CoronaKeyPair,  // PQ signatures (Quasar Phase II)
    pub bls: BlsKeyPair,            // Classical fast-path (Quasar Phase I)
}

impl ValidatorIdentity {
    pub fn new(id: &str) -> Self {
        Self {
            id: id.to_string(),
            kem: KemKeyPair::generate(),
            corona: CoronaKeyPair::generate(),
            bls: BlsKeyPair::generate(),
        }
    }

    pub fn public_key_hash(&self) -> [u8; 32] {
        let mut data = Vec::new();
        data.extend_from_slice(&self.kem.encapsulation_key);
        data.extend_from_slice(&self.corona.verifying_key);
        data.extend_from_slice(&self.bls.public_key);
        *blake3::hash(&data).as_bytes()
    }
}

// ============================================================================
// ZAP Types
// ============================================================================

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
pub enum Effect {
    Pure,
    Deterministic,
    Nondeterministic,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
pub enum Scope {
    Span, File, Repo, Workspace, Node, Chain, Global,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
pub enum TaskState {
    Pending, Running, Completed, Failed, Cancelled,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq)]
pub enum WitnessLevel {
    None, Minimal, Full,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ToolId {
    pub namespace: String,
    pub name: String,
    pub version: String,
}

impl ToolId {
    pub fn native(name: &str) -> Self {
        Self {
            namespace: "native".to_string(),
            name: name.to_string(),
            version: ZAP_VERSION.to_string(),
        }
    }
}

// ============================================================================
// Lux Quasar Consensus Types
// ============================================================================

/// Quasar consensus parameters (from lux/consensus/config)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QuasarConfig {
    pub k: u32,                   // Validators to sample per round
    pub alpha: f64,               // Vote ratio threshold (0.69 mainnet)
    pub alpha_preference: u32,    // β₁ - preference threshold
    pub alpha_confidence: u32,    // β₂ - confidence/finality threshold
    pub beta_virtuous: u32,       // Virtuous confidence count
    pub beta_rogue: u32,          // Rogue/Byzantine confidence count
    pub concurrent_polls: u32,    // Parallel polling
    pub max_rounds: u32,          // Max consensus rounds
    pub round_timeout_ms: u64,    // Round timeout
}

impl QuasarConfig {
    /// Local dev configuration (5 validators)
    pub fn local() -> Self {
        Self {
            k: 5,
            alpha: 0.80,
            alpha_preference: 4,
            alpha_confidence: 4,
            beta_virtuous: 4,
            beta_rogue: 5,
            concurrent_polls: 1,
            max_rounds: 20,
            round_timeout_ms: 5000,
        }
    }

    /// Mainnet configuration (21 validators)
    #[allow(dead_code)]
    pub fn mainnet() -> Self {
        Self {
            k: 21,
            alpha: 0.69,
            alpha_preference: 15,
            alpha_confidence: 15,
            beta_virtuous: 15,
            beta_rogue: 20,
            concurrent_polls: 4,
            max_rounds: 100,
            round_timeout_ms: 10000,
        }
    }
}

/// Photon: a vote message emitted during sampling
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PhotonVote {
    pub round: u32,
    pub peer_id: String,
    pub vote: Vec<u8>,           // Hash of voted proposal
    pub luminance: f64,          // Node brightness/quality (lux units)
    pub bls_signature: Vec<u8>,  // BLS fast-path signature
    pub timestamp: u64,
}

/// Wave: threshold voting state
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WaveState {
    pub proposal_hash: Vec<u8>,
    pub confidence: u32,         // Focus confidence counter
    pub decided: bool,
    pub votes_for: u32,
    pub votes_against: u32,
}

/// Dual certificate: BLS aggregate (classical) + Corona (PQ)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CertBundle {
    pub bls_aggregate: Vec<u8>,      // BLS aggregate signature (96 bytes)
    pub corona_cert: Vec<u8>,      // ML-DSA threshold signature (~3KB)
    pub epoch: u64,                   // Corona epoch number
    pub validator_set: Vec<String>,   // Attesting validators
    pub bls_signers: Vec<String>,     // BLS signers
    pub corona_signers: Vec<String>,// Corona signers
}

impl CertBundle {
    /// Both BLS AND Corona must be valid for quantum finality
    pub fn is_quantum_final(&self) -> bool {
        !self.bls_aggregate.is_empty() && !self.corona_cert.is_empty()
    }
}

/// Quasar block with dual certificates
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QuasarBlock {
    pub height: u64,
    pub hash: Vec<u8>,
    pub parent_hash: Vec<u8>,
    pub state_root: Vec<u8>,
    pub timestamp: u64,
    pub cert: Option<CertBundle>,
}

/// Full Quasar consensus result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QuasarResult {
    pub winner: Vec<u8>,
    pub synthesis: String,
    pub confidence: u32,         // Focus confidence score (β count)
    pub final_round: u32,
    pub total_votes: usize,
    pub cert: CertBundle,
    pub duration_ns: u64,
    pub phase1_ns: u64,          // Nova finality time
    pub phase2_ns: u64,          // Quasar PQ overlay time
}

/// ZAP message content
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ZapContent {
    Text(String),
    TaskRequest {
        description: String,
        priority: u8,
        scope: Scope,
        effect: Effect,
        witness: WitnessLevel,
    },
    TaskResponse {
        task_id: String,
        result: String,
        status: TaskState,
    },
    ToolCall {
        tool_id: ToolId,
        args: serde_json::Value,
    },
    ToolResult {
        call_id: String,
        result: serde_json::Value,
        effect: Effect,
    },
    Ping { nonce: u64 },
    Pong { nonce: u64, latency_ns: u64 },

    /// Quasar: Coordinator proposes a question to the committee
    QuasarPropose {
        question: String,
        config: QuasarConfig,
    },
    /// Quasar: Photon vote from a validator
    QuasarPhoton {
        question_id: u64,
        vote: PhotonVote,
    },
    /// Quasar: Finalized result with dual cert
    QuasarFinalized {
        result: QuasarResult,
    },
}

/// Zero-copy ZAP message with PQ signature
#[derive(Clone)]
pub struct ZapMessage {
    pub id: u64,
    pub from: String,
    pub to: String,
    pub content: ZapContent,
    pub timestamp_ns: u64,
    pub trace_id: u64,
    pub signature: Option<[u8; 4627]>,  // Corona signature
}

impl ZapMessage {
    pub fn new(from: &str, to: &str, content: ZapContent) -> Self {
        Self {
            id: rand::random(),
            from: from.to_string(),
            to: to.to_string(),
            content,
            timestamp_ns: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos() as u64,
            trace_id: rand::random(),
            signature: None,
        }
    }

    pub fn sign(&mut self, identity: &ValidatorIdentity) {
        let data = self.signing_data();
        self.signature = Some(identity.corona.sign(&data));
    }

    fn signing_data(&self) -> Vec<u8> {
        let mut data = Vec::with_capacity(256);
        data.extend_from_slice(&self.id.to_le_bytes());
        data.extend_from_slice(self.from.as_bytes());
        data.extend_from_slice(self.to.as_bytes());
        data.extend_from_slice(&self.timestamp_ns.to_le_bytes());
        data
    }
}

// ============================================================================
// Mesh Coordinator (hanzod)
// ============================================================================

pub struct MeshMetrics {
    pub messages_routed: AtomicU64,
    pub total_latency_ns: AtomicU64,
    pub peak_throughput: AtomicU64,
}

impl MeshMetrics {
    pub fn new() -> Self {
        Self {
            messages_routed: AtomicU64::new(0),
            total_latency_ns: AtomicU64::new(0),
            peak_throughput: AtomicU64::new(0),
        }
    }

    pub fn record(&self, latency_ns: u64) {
        self.messages_routed.fetch_add(1, Ordering::Relaxed);
        self.total_latency_ns.fetch_add(latency_ns, Ordering::Relaxed);
    }

    pub fn avg_latency_ns(&self) -> u64 {
        let count = self.messages_routed.load(Ordering::Relaxed);
        if count == 0 { 0 } else {
            self.total_latency_ns.load(Ordering::Relaxed) / count
        }
    }
}

pub struct MeshPeer {
    pub identity: ValidatorIdentity,
    pub sender: mpsc::Sender<ZapMessage>,
    pub load: AtomicU64,
}

/// hanzod mesh coordinator - Rust-native Lux node
pub struct NativeMesh {
    pub identity: ValidatorIdentity,
    peers: Arc<RwLock<HashMap<String, Arc<MeshPeer>>>>,
    broadcast: broadcast::Sender<ZapMessage>,
    metrics: Arc<MeshMetrics>,
    message_log: Arc<RwLock<Vec<ZapMessage>>>,
}

impl NativeMesh {
    pub fn new(id: &str) -> Self {
        let (broadcast, _) = broadcast::channel(1024);
        Self {
            identity: ValidatorIdentity::new(id),
            peers: Arc::new(RwLock::new(HashMap::new())),
            broadcast,
            metrics: Arc::new(MeshMetrics::new()),
            message_log: Arc::new(RwLock::new(Vec::new())),
        }
    }

    pub async fn register(&self, identity: ValidatorIdentity) -> mpsc::Receiver<ZapMessage> {
        let (tx, rx) = mpsc::channel(256);
        let peer = Arc::new(MeshPeer {
            identity: identity.clone(),
            sender: tx,
            load: AtomicU64::new(0),
        });

        let mut peers = self.peers.write().await;
        peers.insert(identity.id.clone(), peer);

        println!("  [+] Validator '{}' registered (BLS + Corona + ML-KEM-1024)", identity.id);
        rx
    }

    pub async fn route(&self, msg: ZapMessage) -> Result<(), String> {
        let start = Instant::now();

        {
            let mut log = self.message_log.write().await;
            log.push(msg.clone());
        }

        let peers = self.peers.read().await;
        if let Some(peer) = peers.get(&msg.to) {
            peer.sender.send(msg.clone()).await.map_err(|e| e.to_string())?;
            peer.load.fetch_add(1, Ordering::Relaxed);
            self.metrics.record(start.elapsed().as_nanos() as u64);
            Ok(())
        } else {
            Err(format!("Peer '{}' not found", msg.to))
        }
    }

    pub async fn broadcast(&self, msg: ZapMessage) -> usize {
        let _ = self.broadcast.send(msg);
        self.broadcast.receiver_count()
    }

    pub fn metrics(&self) -> &MeshMetrics {
        &self.metrics
    }

    pub async fn message_count(&self) -> usize {
        self.message_log.read().await.len()
    }

    pub async fn peer_count(&self) -> usize {
        self.peers.read().await.len()
    }
}

// ============================================================================
// Native ZAP Agent (Hanzo Validator)
// ============================================================================

pub struct NativeAgent {
    pub identity: ValidatorIdentity,
    pub name: String,
    inbox: mpsc::Receiver<ZapMessage>,
    mesh: Arc<NativeMesh>,
}

impl NativeAgent {
    pub async fn connect(mesh: Arc<NativeMesh>, name: &str) -> Self {
        let identity = ValidatorIdentity::new(&format!("agent-{}", name.to_lowercase()));
        let inbox = mesh.register(identity.clone()).await;

        Self {
            identity,
            name: name.to_string(),
            inbox,
            mesh,
        }
    }

    pub fn id(&self) -> &str {
        &self.identity.id
    }

    pub async fn send(&self, to: &str, content: ZapContent) -> Result<(), String> {
        let mut msg = ZapMessage::new(&self.identity.id, to, content);
        msg.sign(&self.identity);
        self.mesh.route(msg).await
    }

    pub async fn recv(&mut self, timeout_ms: u64) -> Option<ZapMessage> {
        tokio::time::timeout(
            Duration::from_millis(timeout_ms),
            self.inbox.recv()
        ).await.ok().flatten()
    }

    pub async fn ping(&self, to: &str) -> Result<u64, String> {
        let nonce: u64 = rand::random();
        let start = Instant::now();
        self.send(to, ZapContent::Ping { nonce }).await?;
        Ok(start.elapsed().as_nanos() as u64)
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_native_pq_handshake() {
        println!("\n  Hanzo E2E: PQ Handshake (ML-KEM-1024 + ML-DSA-87 + BLS)\n");

        let mesh = Arc::new(NativeMesh::new("hanzod-local"));

        let agent_a = NativeAgent::connect(mesh.clone(), "Alice").await;
        let mut agent_b = NativeAgent::connect(mesh.clone(), "Bob").await;

        let bob_id = agent_b.id().to_string();

        agent_a.send(&bob_id, ZapContent::Text("Hello from Alice with Corona signature!".to_string())).await.unwrap();
        tokio::time::sleep(Duration::from_millis(10)).await;

        let msg = agent_b.recv(1000).await;
        assert!(msg.is_some());
        let msg = msg.unwrap();
        assert!(msg.signature.is_some(), "Message should be Corona-signed");
        println!("  [ok] Bob received Corona-signed message (sig: {:x?}...)", &msg.signature.unwrap()[..16]);
        println!("  [ok] PQ Handshake passed\n");
    }

    #[tokio::test]
    async fn test_native_high_throughput() {
        println!("\n  Hanzo E2E: High Throughput (ZAP over hanzod)\n");

        let mesh = Arc::new(NativeMesh::new("hanzod-perf"));

        let mut agents: Vec<NativeAgent> = Vec::new();
        for i in 0..10 {
            agents.push(NativeAgent::connect(mesh.clone(), &format!("V-{i}")).await);
        }
        let ids: Vec<String> = agents.iter().map(|a| a.id().to_string()).collect();

        let start = Instant::now();
        let num_messages = 1000;

        for round in 0..num_messages {
            let from_idx = round % agents.len();
            let to_idx = (round + 1) % agents.len();
            agents[from_idx].send(&ids[to_idx], ZapContent::Text(format!("msg-{round}"))).await.unwrap();
        }

        let elapsed = start.elapsed();
        let throughput = num_messages as f64 / elapsed.as_secs_f64();
        tokio::time::sleep(Duration::from_millis(100)).await;

        println!("  Messages:   {num_messages}");
        println!("  Duration:   {elapsed:?}");
        println!("  Throughput: {throughput:.0} msg/sec");
        println!("  Avg lat:    {} ns", mesh.metrics().avg_latency_ns());

        assert!(throughput > 1000.0, "Throughput should exceed 1000 msg/sec");
        println!("  [ok] High Throughput passed ({throughput:.0} msg/sec)\n");
    }

    #[tokio::test]
    async fn test_native_task_workflow() {
        println!("\n  Hanzo E2E: Task Workflow (CTO -> Dev -> Reviewer)\n");

        let mesh = Arc::new(NativeMesh::new("hanzod-workflow"));

        let cto = NativeAgent::connect(mesh.clone(), "CTO").await;
        let mut dev = NativeAgent::connect(mesh.clone(), "Developer").await;
        let mut reviewer = NativeAgent::connect(mesh.clone(), "Reviewer").await;

        let dev_id = dev.id().to_string();
        let reviewer_id = reviewer.id().to_string();
        let cto_id = cto.id().to_string();

        cto.send(&dev_id, ZapContent::TaskRequest {
            description: "Implement Quasar wire protocol in Rust".to_string(),
            priority: 1,
            scope: Scope::Repo,
            effect: Effect::Nondeterministic,
            witness: WitnessLevel::Full,
        }).await.unwrap();

        tokio::time::sleep(Duration::from_millis(20)).await;
        let task = dev.recv(1000).await.unwrap();

        dev.send(&reviewer_id, ZapContent::TaskRequest {
            description: "Review Quasar wire protocol implementation".to_string(),
            priority: 1,
            scope: Scope::File,
            effect: Effect::Pure,
            witness: WitnessLevel::Minimal,
        }).await.unwrap();

        tokio::time::sleep(Duration::from_millis(20)).await;
        let review_req = reviewer.recv(1000).await.unwrap();

        reviewer.send(&dev_id, ZapContent::TaskResponse {
            task_id: format!("{}", review_req.id),
            result: "LGTM - Approved".to_string(),
            status: TaskState::Completed,
        }).await.unwrap();

        tokio::time::sleep(Duration::from_millis(20)).await;
        let _approval = dev.recv(1000).await.unwrap();

        dev.send(&cto_id, ZapContent::TaskResponse {
            task_id: format!("{}", task.id),
            result: "Quasar protocol implemented and reviewed".to_string(),
            status: TaskState::Completed,
        }).await.unwrap();

        assert_eq!(mesh.message_count().await, 4);
        println!("  [ok] Task Workflow passed (4 messages routed)\n");
    }

    #[tokio::test]
    async fn test_native_tool_invocation() {
        println!("\n  Hanzo E2E: Tool Invocation (ZAP native.fs.read)\n");

        let mesh = Arc::new(NativeMesh::new("hanzod-tools"));

        let client = NativeAgent::connect(mesh.clone(), "Client").await;
        let mut provider = NativeAgent::connect(mesh.clone(), "ToolProvider").await;

        let provider_id = provider.id().to_string();
        let client_id = client.id().to_string();

        client.send(&provider_id, ZapContent::ToolCall {
            tool_id: ToolId::native("fs.read"),
            args: serde_json::json!({ "path": "/etc/hosts", "limit": 10 }),
        }).await.unwrap();

        tokio::time::sleep(Duration::from_millis(20)).await;
        let call = provider.recv(1000).await.unwrap();
        match &call.content {
            ZapContent::ToolCall { tool_id, args } => {
                println!("  Provider received: {}.{} v{}", tool_id.namespace, tool_id.name, tool_id.version);
                println!("  Args: {args}");
            }
            _ => panic!("Expected ToolCall"),
        }

        provider.send(&client_id, ZapContent::ToolResult {
            call_id: format!("{}", call.id),
            result: serde_json::json!({ "content": "127.0.0.1 localhost", "lines": 1 }),
            effect: Effect::Deterministic,
        }).await.unwrap();

        println!("  [ok] Tool Invocation passed\n");
    }

    #[tokio::test]
    async fn test_native_concurrent_stress() {
        println!("\n  Hanzo E2E: Concurrent Stress (50 validators)\n");

        let mesh = Arc::new(NativeMesh::new("hanzod-stress"));
        let num_agents = 50;
        let messages_per_agent = 100;

        let mut agent_ids = Vec::new();
        for i in 0..num_agents {
            let agent = NativeAgent::connect(mesh.clone(), &format!("V-{i}")).await;
            agent_ids.push(agent.id().to_string());
        }

        let start = Instant::now();
        let mut handles = Vec::new();

        for i in 0..num_agents {
            let mesh_clone = mesh.clone();
            let ids = agent_ids.clone();
            let from_id = ids[i].clone();

            let handle = tokio::spawn(async move {
                let identity = ValidatorIdentity::new(&from_id);
                for j in 0..messages_per_agent {
                    let to_idx = (i + j + 1) % num_agents;
                    let mut msg = ZapMessage::new(
                        &from_id,
                        &ids[to_idx],
                        ZapContent::Ping { nonce: j as u64 },
                    );
                    msg.sign(&identity);
                    let _ = mesh_clone.route(msg).await;
                }
            });
            handles.push(handle);
        }

        for h in handles {
            h.await.unwrap();
        }

        let elapsed = start.elapsed();
        let total_messages = num_agents * messages_per_agent;
        let throughput = total_messages as f64 / elapsed.as_secs_f64();

        println!("  Agents:     {num_agents}");
        println!("  Messages:   {total_messages}");
        println!("  Duration:   {elapsed:?}");
        println!("  Throughput: {throughput:.0} msg/sec");

        assert!(throughput > 10000.0, "Concurrent throughput should exceed 10k msg/sec");
        println!("  [ok] Concurrent Stress passed ({throughput:.0} msg/sec)\n");
    }

    /// Quasar Consensus: N-round voting with Photon sampling, Wave thresholds,
    /// Focus confidence, dual BLS + Corona certificate generation.
    ///
    /// All validators MUST participate in every round (verified by participation bitmap).
    #[tokio::test]
    async fn test_quasar_consensus() {
        println!("\n  Hanzo E2E: Quasar Consensus (Lux PQ Protocol)\n");
        println!("  ================================================================");

        let mesh = Arc::new(NativeMesh::new("hanzod-quasar"));
        let question = "What is the best approach for implementing post-quantum secure agent communication?";
        let config = QuasarConfig::local();

        println!("  Question: \"{question}\"");
        println!();
        println!("  Quasar Config (local):");
        println!("    k: {}  alpha: {:.0}%  alpha_pref(B1): {}  alpha_conf(B2): {}",
                 config.k, config.alpha * 100.0, config.alpha_preference, config.alpha_confidence);
        println!("    beta_virtuous: {}  beta_rogue: {}  max_rounds: {}",
                 config.beta_virtuous, config.beta_rogue, config.max_rounds);
        println!();

        // 3 candidate proposals
        let proposals = [
            "ML-KEM-1024 + ML-DSA-87 hybrid with X25519 fallback (FIPS 203/204)",
            "ZAP PQ envelope: ML-KEM-1024 KEM, AES-256-GCM payload, ML-DSA-87 auth",
            "Full PQ stack: ML-KEM-1024 key exchange, ML-DSA-87 sigs, SPHINCS+ backup",
        ];
        let proposal_hashes: Vec<Vec<u8>> = proposals.iter()
            .map(|p| blake3::hash(p.as_bytes()).as_bytes().to_vec())
            .collect();

        println!("  Proposals:");
        for (i, p) in proposals.iter().enumerate() {
            println!("    [{i}] \"{p}\"");
        }
        println!();

        // Spawn 5 validators
        let validator_names = ["Cryptographer", "Architect", "Auditor", "Engineer", "Validator"];
        let num_validators = validator_names.len();
        let mut validators: Vec<NativeAgent> = Vec::new();
        for name in &validator_names {
            validators.push(NativeAgent::connect(mesh.clone(), name).await);
        }
        let validator_ids: Vec<String> = validators.iter().map(|v| v.id().to_string()).collect();
        println!();

        // Coordinator broadcasts Quasar proposal
        let coordinator_id = "hanzod-quasar".to_string();
        for id in &validator_ids {
            let msg = ZapMessage::new(&coordinator_id, id, ZapContent::QuasarPropose {
                question: question.to_string(),
                config: config.clone(),
            });
            mesh.route(msg).await.unwrap();
        }
        tokio::time::sleep(Duration::from_millis(30)).await;

        // Each validator receives the proposal
        for v in &mut validators {
            let msg = v.recv(1000).await.unwrap();
            match &msg.content {
                ZapContent::QuasarPropose { question: q, .. } => assert_eq!(q, question),
                _ => panic!("Expected QuasarPropose"),
            }
        }

        // ================================================================
        // Quasar N-Round Consensus
        //
        // Phase I (Nova): Photon sampling + Wave threshold + Focus confidence
        // Phase II (Quasar): Dual BLS + Corona certificate
        // ================================================================

        let start = Instant::now();
        let phase1_start = start;

        // Initial preferences: spread across proposals
        let initial_prefs = [0usize, 0, 1, 1, 2];
        let mut preferences: Vec<usize> = initial_prefs.to_vec();
        let mut luminance: Vec<f64> = vec![800.0, 750.0, 820.0, 780.0, 700.0]; // lux units
        let mut focus_confidence: Vec<u32> = vec![0; num_validators]; // Focus counter per proposal
        let mut consecutive_decided: u32 = 0;
        let mut finalized = false;
        let mut final_round = 0u32;

        // Per-round tracking
        let mut round_history: Vec<Vec<usize>> = Vec::new();
        let mut participation: Vec<Vec<bool>> = Vec::new();
        let mut all_photons: Vec<PhotonVote> = Vec::new();

        println!("  ---- Photon Sampling + Wave Voting + Focus Confidence ----\n");
        print!("  Round  0 (initial): ");
        for (i, name) in validator_names.iter().enumerate() {
            print!("{}:[{}]  ", &name[..4], preferences[i]);
        }
        println!("\n");

        for round in 1..=config.max_rounds {
            let mut round_participated = vec![false; num_validators];
            let mut round_prefs = preferences.clone();

            for agent_idx in 0..num_validators {
                round_participated[agent_idx] = true;

                // --- Photon: sample k peers (excluding self) ---
                let mut peer_indices: Vec<usize> = (0..num_validators)
                    .filter(|&j| j != agent_idx)
                    .collect();
                let rotation = ((round as usize) * 7 + agent_idx * 3) % peer_indices.len();
                peer_indices.rotate_left(rotation);
                peer_indices.truncate(config.k.min(peer_indices.len() as u32) as usize);

                // Collect sampled preferences (weighted by luminance)
                let mut proposal_weight = vec![0.0f64; proposals.len()];
                for &peer_idx in &peer_indices {
                    proposal_weight[preferences[peer_idx]] += luminance[peer_idx];
                }

                let total_weight: f64 = proposal_weight.iter().sum();
                let (majority_proposal, majority_weight) = proposal_weight.iter()
                    .enumerate()
                    .max_by(|(_, a), (_, b)| a.partial_cmp(b).unwrap())
                    .unwrap();

                let wave_ratio = majority_weight / total_weight;

                // --- Wave: threshold voting ---
                if wave_ratio >= config.alpha {
                    round_prefs[agent_idx] = majority_proposal;
                    // --- Focus: increment confidence ---
                    focus_confidence[agent_idx] += 1;
                    luminance[agent_idx] = (luminance[agent_idx] * 1.02).min(1000.0);
                } else {
                    // Rogue path: partial influence
                    let flip = ((round as f64 * 0.37 + agent_idx as f64 * 0.53) % 1.0) < wave_ratio;
                    if flip && majority_proposal != preferences[agent_idx] {
                        round_prefs[agent_idx] = majority_proposal;
                        focus_confidence[agent_idx] = 0; // Reset on switch
                    }
                }

                // Record Photon vote with BLS signature
                let vote_hash = proposal_hashes[round_prefs[agent_idx]].clone();
                let vote_data = format!("{round}:{}", validator_ids[agent_idx]);
                let bls_sig = validators[agent_idx].identity.bls.sign(vote_data.as_bytes());

                all_photons.push(PhotonVote {
                    round,
                    peer_id: validator_ids[agent_idx].clone(),
                    vote: vote_hash,
                    luminance: luminance[agent_idx],
                    bls_signature: bls_sig.to_vec(),
                    timestamp: std::time::SystemTime::now()
                        .duration_since(std::time::UNIX_EPOCH)
                        .unwrap()
                        .as_nanos() as u64,
                });
            }

            // Apply round updates
            preferences = round_prefs;
            round_history.push(preferences.clone());
            participation.push(round_participated.clone());

            // Check supermajority
            let mut proposal_counts = vec![0usize; proposals.len()];
            for &pref in &preferences {
                proposal_counts[pref] += 1;
            }
            let max_count = *proposal_counts.iter().max().unwrap();
            let supermajority = max_count as f64 / num_validators as f64;

            // Print round
            let all_participated = round_participated.iter().all(|&p| p);
            let marker = if all_participated { "ok" } else { "!!" };

            print!("  Round {:2} [{marker} {num_validators}/{num_validators}] ", round);
            for (i, _) in validator_names.iter().enumerate() {
                let changed = if round > 1 && preferences[i] != round_history[round as usize - 2][i] {
                    ">"
                } else {
                    " "
                };
                print!("[{}]{}{:3}  ", preferences[i], changed, focus_confidence[i]);
            }
            print!(" majority: {:.0}% ({max_count})", supermajority * 100.0);

            if supermajority >= config.alpha {
                consecutive_decided += 1;
                print!("  streak:{consecutive_decided}");
            } else {
                consecutive_decided = 0;
            }
            println!();

            // Focus finality: β_virtuous consecutive rounds with supermajority
            if consecutive_decided >= config.beta_virtuous {
                final_round = round;
                finalized = true;
                println!("\n  FINALIZED at round {round} (beta_virtuous={} consecutive supermajority rounds)", config.beta_virtuous);
                break;
            }

            // Unanimous early exit
            if max_count == num_validators && consecutive_decided >= config.alpha_preference {
                final_round = round;
                finalized = true;
                println!("\n  UNANIMOUS at round {round} (all {num_validators} validators, streak {consecutive_decided})");
                break;
            }
        }

        if !finalized {
            final_round = config.max_rounds;
            println!("\n  Max rounds reached ({})", config.max_rounds);
        }

        let phase1_ns = phase1_start.elapsed().as_nanos() as u64;

        // ================================================================
        // Phase II: Quasar PQ Overlay - Dual Certificate Generation
        // ================================================================

        println!("\n  ---- Quasar Phase II: Dual Certificate (BLS + Corona) ----\n");

        let phase2_start = Instant::now();

        let winning_proposal = preferences[0];
        let winning_text = proposals[winning_proposal];
        let winning_hash = &proposal_hashes[winning_proposal];

        // BLS aggregate: collect all BLS signatures for final round
        let final_photons: Vec<&PhotonVote> = all_photons.iter()
            .filter(|p| p.round == final_round)
            .collect();

        let mut bls_aggregate = vec![0u8; 96];
        let mut bls_signers = Vec::new();
        for photon in &final_photons {
            // XOR aggregate (real: BLS pairing-based aggregation)
            for (i, b) in photon.bls_signature.iter().enumerate().take(96) {
                bls_aggregate[i] ^= b;
            }
            bls_signers.push(photon.peer_id.clone());
        }
        println!("  BLS aggregate: {:x?}... ({} signers)", &bls_aggregate[..16], bls_signers.len());

        // Corona threshold: each validator signs with ML-DSA-87
        let cert_data = format!("{final_round}:{}", hex::encode(winning_hash));
        let mut corona_cert = Vec::new();
        let mut corona_signers = Vec::new();
        for v in &validators {
            let sig = v.identity.corona.sign(cert_data.as_bytes());
            corona_cert.extend_from_slice(&sig[..64]); // Truncated for display
            corona_signers.push(v.id().to_string());
        }
        println!("  Corona cert: {:x?}... ({} signers, {}B)", &corona_cert[..16], corona_signers.len(), corona_cert.len());

        let cert = CertBundle {
            bls_aggregate: bls_aggregate.clone(),
            corona_cert,
            epoch: 1,
            validator_set: validator_ids.clone(),
            bls_signers,
            corona_signers,
        };

        assert!(cert.is_quantum_final(), "Both BLS and Corona must be present");
        println!("  Quantum finality: {}", cert.is_quantum_final());

        let phase2_ns = phase2_start.elapsed().as_nanos() as u64;
        let duration_ns = start.elapsed().as_nanos() as u64;

        // ================================================================
        // Participation Analysis
        // ================================================================

        let total_rounds = round_history.len();

        println!("\n  ---- Participation ----\n");
        for (i, name) in validator_names.iter().enumerate() {
            let rounds_participated: usize = participation.iter()
                .map(|r| if r[i] { 1 } else { 0 })
                .sum();
            let pct = rounds_participated as f64 / total_rounds as f64 * 100.0;
            let ok = if rounds_participated == total_rounds { "ok" } else { "FAIL" };
            println!("  [{ok}] {name}: {rounds_participated}/{total_rounds} rounds ({pct:.0}%)");
            assert_eq!(rounds_participated, total_rounds,
                       "Validator '{name}' must participate in ALL rounds");
        }

        // ================================================================
        // Convergence Analysis
        // ================================================================

        println!("\n  ---- Convergence ----\n");
        for (i, name) in validator_names.iter().enumerate() {
            let first_adopted = round_history.iter()
                .position(|r| r[i] == winning_proposal)
                .map(|r| r + 1)
                .unwrap_or(0);
            let switched = initial_prefs[i] != winning_proposal;
            println!("  {name}: initial=[{}] {} winner=[{winning_proposal}] at round {first_adopted}",
                     initial_prefs[i], if switched { ">" } else { "=" });
        }

        // ================================================================
        // Build Result
        // ================================================================

        let overall_confidence = focus_confidence.iter().sum::<u32>() / num_validators as u32;

        let result = QuasarResult {
            winner: winning_hash.clone(),
            synthesis: format!("QUASAR ({num_validators}/{num_validators} at round {final_round}): {winning_text}"),
            confidence: overall_confidence,
            final_round,
            total_votes: all_photons.len(),
            cert: cert.clone(),
            duration_ns,
            phase1_ns,
            phase2_ns,
        };

        // ================================================================
        // Print Final Result
        // ================================================================

        println!();
        println!("  ================================================================");
        println!("  QUASAR CONSENSUS RESULT");
        println!("  ================================================================");
        println!();
        println!("  Question: \"{question}\"");
        println!();
        println!("  Answer [{winning_proposal}]: \"{winning_text}\"");
        println!();
        println!("  Metrics:");
        println!("    Focus confidence:  {}", result.confidence);
        println!("    Final round:       {final_round} / {} max", config.max_rounds);
        println!("    Total Photon votes:{} ({num_validators} validators x {total_rounds} rounds)", all_photons.len());
        println!("    Finalized:         {finalized}");
        println!("    Phase I  (Nova):   {:.2}ms", phase1_ns as f64 / 1_000_000.0);
        println!("    Phase II (Quasar): {:.2}ms", phase2_ns as f64 / 1_000_000.0);
        println!("    Total:             {:.2}ms", duration_ns as f64 / 1_000_000.0);
        println!();
        println!("  CertBundle:");
        println!("    bls_aggregate:  {:x?}... (96B)", &cert.bls_aggregate[..16]);
        println!("    corona_cert:  {:x?}... ({}B)", &cert.corona_cert[..16], cert.corona_cert.len());
        println!("    epoch:          {}", cert.epoch);
        println!("    quantum_final:  {}", cert.is_quantum_final());
        println!("    validators:     {}", cert.validator_set.len());
        for v in &cert.validator_set {
            println!("      - {v}");
        }
        println!();
        println!("  Crypto: BLS12-381 (Phase I) + ML-DSA-87/Corona (Phase II)");
        println!("  PQ KEM: ML-KEM-1024 (FIPS 203)");
        println!("  ZAP:    v{ZAP_VERSION} ({ZAP_PROFILE})");
        println!("  Quasar: v{QUASAR_VERSION}");
        println!();
        println!("  ================================================================");

        // ================================================================
        // Assertions
        // ================================================================

        for round_p in &participation {
            assert!(round_p.iter().all(|&p| p), "All validators must participate in every round");
        }

        assert!(finalized, "Quasar consensus should finalize within {} rounds", config.max_rounds);

        let all_agree = preferences.iter().all(|&p| p == winning_proposal);
        assert!(all_agree, "All validators should converge to winning proposal");

        assert!(cert.is_quantum_final(), "CertBundle needs both BLS and Corona");
        assert_eq!(cert.validator_set.len(), num_validators);
        assert_eq!(all_photons.len(), num_validators * total_rounds);

        println!("\n  [ok] Quasar Consensus passed ({total_rounds} rounds, {} Photon votes)\n", all_photons.len());
    }
}