cp_validator/

evaluation.rs

//! Evaluation pipeline for content contributors and live peers.
//!
//! Per CP-015 section 5: Validators download content from Arweave, index it
//! locally in an isolated substrate, run test queries, and measure retrieval
//! quality using precision@10, NDCG@10, and MRR.
//!
//! Per CP-015 section 13: Validators test live peers by connecting over Tor,
//! sending test queries, and measuring search quality and latency.

use crate::corpus::TestQuery;
use crate::rating::Rating;
use crate::window::ValidationWindow;
use cp_arweave::ArweaveClient;
use cp_graph::GraphStore;
use cp_tor::types::{PeerRegistration, SearchResponse, SearchStatus};
use ed25519_dalek::{Signer, SigningKey};
use serde::{Deserialize, Serialize};
use serde_big_array::BigArray;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Instant;
use tracing::{debug, info, warn};
use uuid::Uuid;

/// Quality metrics for a contributor's content, measured against ground truth.
///
/// Per CP-015 section 1.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QualityMetrics {
    /// Fraction of top-10 results that are relevant.
    pub precision_at_10: f32,
    /// Normalized discounted cumulative gain at 10.
    pub ndcg_at_10: f32,
    /// Mean reciprocal rank of first relevant result.
    pub mrr: f32,
    /// Number of valid Merkle proofs in content.
    pub merkle_proofs_valid: u32,
    /// Number of invalid or missing proofs.
    pub merkle_proofs_invalid: u32,
    /// Adapter perplexity delta (negative means improvement).
    pub adapter_perplexity_delta: Option<f32>,
}

/// Result of evaluating a single contributor's content.
///
/// Per CP-015 section 10.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EvaluationResult {
    /// Ed25519 public key of the contributor.
    pub contributor_key: [u8; 32],
    /// Node ID of the validator that performed this evaluation.
    pub validator_node_id: [u8; 16],
    /// Window ID when this evaluation was performed.
    pub window_id: u64,
    /// Aggregated quality metrics.
    pub metrics: QualityMetrics,
    /// Timestamp of the evaluation (Unix milliseconds).
    pub timestamp: i64,
    /// Ed25519 signature over (`contributor_key` || `window_id` || metrics).
    #[serde(with = "BigArray")]
    pub signature: [u8; 64],
}

impl EvaluationResult {
    /// Compute the bytes to sign: `BLAKE3(CBOR(contributor_key` || `window_id` || metrics)).
    pub fn signing_bytes(&self) -> [u8; 32] {
        let signable = EvaluationResultSignable {
            contributor_key: &self.contributor_key,
            window_id: self.window_id,
            metrics: &self.metrics,
            timestamp: self.timestamp,
        };
        let mut buf = Vec::new();
        ciborium::into_writer(&signable, &mut buf).expect("CBOR serialization cannot fail");
        *blake3::hash(&buf).as_bytes()
    }
}

#[derive(Serialize)]
struct EvaluationResultSignable<'a> {
    contributor_key: &'a [u8; 32],
    window_id: u64,
    metrics: &'a QualityMetrics,
    timestamp: i64,
}

/// Quality metrics for a live peer, including latency and availability.
///
/// Per CP-015 section 13.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerQualityMetrics {
    /// Fraction of top-10 results that are relevant.
    pub precision_at_10: f32,
    /// Normalized discounted cumulative gain at 10.
    pub ndcg_at_10: f32,
    /// Mean reciprocal rank of first relevant result.
    pub mrr: f32,
    /// Average response latency in milliseconds.
    pub response_latency_ms: u16,
    /// Fraction of results with valid Merkle proofs.
    pub proof_validity_rate: f32,
    /// Fraction of test windows where peer was reachable.
    pub availability_rate: f32,
}

/// Result of testing a live peer.
///
/// Per CP-015 section 13.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerTestResult {
    /// Node ID of the tested peer.
    pub peer_node_id: [u8; 16],
    /// Onion address of the tested peer.
    pub peer_onion: String,
    /// Node ID of the validator.
    pub validator_node_id: [u8; 16],
    /// Window ID when this test was performed.
    pub window_id: u64,
    /// Quality metrics.
    pub metrics: PeerQualityMetrics,
    /// Timestamp (Unix milliseconds).
    pub timestamp: i64,
    /// Ed25519 signature.
    #[serde(with = "BigArray")]
    pub signature: [u8; 64],
}

impl PeerTestResult {
    /// Compute the bytes to sign.
    pub fn signing_bytes(&self) -> [u8; 32] {
        let signable = PeerTestResultSignable {
            peer_node_id: &self.peer_node_id,
            peer_onion: &self.peer_onion,
            window_id: self.window_id,
            metrics: &self.metrics,
            timestamp: self.timestamp,
        };
        let mut buf = Vec::new();
        ciborium::into_writer(&signable, &mut buf).expect("CBOR serialization cannot fail");
        *blake3::hash(&buf).as_bytes()
    }
}

#[derive(Serialize)]
struct PeerTestResultSignable<'a> {
    peer_node_id: &'a [u8; 16],
    peer_onion: &'a str,
    window_id: u64,
    metrics: &'a PeerQualityMetrics,
    timestamp: i64,
}

// ============================================================================
// Retrieval quality metrics
// ============================================================================

/// Precision at K: fraction of the top-K results that are relevant.
///
/// Given a ranked list of returned IDs and a set of known relevant IDs,
/// counts how many of the top K returned IDs appear in the relevant set.
///
/// Returns 0.0 if k is 0 or returned is empty.
pub fn precision_at_k(returned: &[Uuid], relevant: &[Uuid], k: usize) -> f32 {
    if k == 0 || returned.is_empty() {
        return 0.0;
    }

    let take = returned.len().min(k);
    let relevant_count = returned[..take]
        .iter()
        .filter(|id| relevant.contains(id))
        .count();

    relevant_count as f32 / k as f32
}

/// Normalized Discounted Cumulative Gain at K.
///
/// Measures ranking quality accounting for position: relevant results
/// appearing earlier are more valuable. Uses graded relevance (0-3)
/// when available, falling back to binary relevance from the relevant set.
///
/// The `relevance_grades` map provides graded relevance for each ID.
/// IDs not in the map but present in the relevant set get grade 1.
/// IDs in neither get grade 0.
///
/// DCG@K = sum_{i=1}^{K} (`2^rel_i` - 1) / log2(i + 1)
/// NDCG@K = DCG@K / IDCG@K
///
/// where IDCG@K is the DCG of the ideal ranking.
pub fn ndcg_at_k(
    returned: &[Uuid],
    relevant: &[Uuid],
    relevance_grades: &HashMap<Uuid, u8>,
    k: usize,
) -> f32 {
    if k == 0 || returned.is_empty() || relevant.is_empty() {
        return 0.0;
    }

    let take = returned.len().min(k);

    // Compute DCG for the returned ranking
    let dcg = compute_dcg(&returned[..take], relevant, relevance_grades);

    // Compute ideal DCG: sort all known relevant items by their grade, descending
    let mut ideal_grades: Vec<u8> = relevant
        .iter()
        .map(|id| *relevance_grades.get(id).unwrap_or(&1))
        .collect();
    ideal_grades.sort_unstable_by(|a, b| b.cmp(a));
    ideal_grades.truncate(k);

    let idcg: f64 = ideal_grades
        .iter()
        .enumerate()
        .map(|(i, &grade)| {
            let gain = (2.0_f64).powi(i32::from(grade)) - 1.0;
            gain / (i as f64 + 2.0).log2()
        })
        .sum();

    if idcg == 0.0 {
        return 0.0;
    }

    (dcg / idcg) as f32
}

/// Compute Discounted Cumulative Gain for a ranked list.
fn compute_dcg(ranked: &[Uuid], relevant: &[Uuid], relevance_grades: &HashMap<Uuid, u8>) -> f64 {
    ranked
        .iter()
        .enumerate()
        .map(|(i, id)| {
            let grade = if let Some(&g) = relevance_grades.get(id) {
                g
            } else {
                u8::from(relevant.contains(id))
            };
            let gain = (2.0_f64).powi(i32::from(grade)) - 1.0;
            gain / (i as f64 + 2.0).log2()
        })
        .sum()
}

/// Mean Reciprocal Rank: the reciprocal of the rank of the first relevant result.
///
/// If no relevant result is found in the returned list, returns 0.0.
/// Rank is 1-indexed (first position = rank 1).
pub fn mrr(returned: &[Uuid], relevant: &[Uuid]) -> f32 {
    for (i, id) in returned.iter().enumerate() {
        if relevant.contains(id) {
            return 1.0 / (i as f32 + 1.0);
        }
    }
    0.0
}

/// Composite quality score for a contributor evaluation result.
///
/// Per CP-015 section 7:
///   score = 0.4 * ndcg@10 + 0.3 * precision@10 + 0.2 * mrr
///         + 0.1 * `proof_validity_rate`
///   With a 0.8 penalty multiplier if `adapter_perplexity_delta` > 0.
pub fn composite_score(metrics: &QualityMetrics) -> f32 {
    let proof_rate = if metrics.merkle_proofs_valid + metrics.merkle_proofs_invalid > 0 {
        metrics.merkle_proofs_valid as f32
            / (metrics.merkle_proofs_valid + metrics.merkle_proofs_invalid) as f32
    } else {
        0.0
    };

    let mut score = 0.4 * metrics.ndcg_at_10
        + 0.3 * metrics.precision_at_10
        + 0.2 * metrics.mrr
        + 0.1 * proof_rate;

    // Penalty for adapter degradation
    if let Some(delta) = metrics.adapter_perplexity_delta {
        if delta > 0.0 {
            score *= 0.8;
        }
    }

    score
}

/// Composite quality score for a peer test result.
///
/// Per CP-015 section 14:
///   70% quality + 20% reliability + 10% latency
pub fn peer_composite_score(metrics: &PeerQualityMetrics) -> f32 {
    let quality = 0.4 * metrics.ndcg_at_10 + 0.2 * metrics.precision_at_10 + 0.1 * metrics.mrr;

    let reliability = metrics.proof_validity_rate * metrics.availability_rate;

    // Latency score: 1.0 at 0ms, 0.0 at 5000ms, linear
    let latency_score = (1.0 - f32::from(metrics.response_latency_ms) / 5000.0).max(0.0);

    0.7 * quality + 0.2 * reliability + 0.1 * latency_score
}

/// Evaluate a contributor by downloading their content from Arweave,
/// indexing it in an isolated substrate, and running test queries.
///
/// Per CP-015 section 5.
pub async fn evaluate_contributor(
    contributor_key: [u8; 32],
    arweave: &ArweaveClient,
    test_queries: &[TestQuery],
    validator_node_id: [u8; 16],
    signing_key: &SigningKey,
) -> Result<EvaluationResult, crate::ValidatorError> {
    info!(
        contributor = hex::encode(contributor_key),
        queries = test_queries.len(),
        "Evaluating contributor"
    );

    // 1. Discover contributor's content on Arweave
    let contributor_hex = hex::encode(contributor_key);
    let (discovered, _cursor) =
        cp_arweave::discover_diffs_by_contributor(arweave, &contributor_hex, 50, None)
            .await
            .map_err(|e| crate::ValidatorError::Arweave(e.to_string()))?;

    if discovered.is_empty() {
        return Err(crate::ValidatorError::NoContent(contributor_hex));
    }

    // 2. Build a temporary local index of the contributor's content
    let temp_dir = tempfile::tempdir()
        .map_err(|e| crate::ValidatorError::Internal(format!("Failed to create temp dir: {e}")))?;
    let db_path = temp_dir.path().join("eval.db");
    let db_path_str = db_path.to_string_lossy().to_string();

    let mut local_index = GraphStore::open(&db_path_str)
        .map_err(|e| crate::ValidatorError::Internal(format!("Failed to open graph store: {e}")))?;

    let mut downloaded_count = 0u64;
    for diff_meta in &discovered {
        match cp_arweave::download_diff(arweave, diff_meta).await {
            Ok(downloaded) => {
                for doc in &downloaded.diff.added_docs {
                    let _ = local_index.insert_document(doc);
                }
                for chunk in &downloaded.diff.added_chunks {
                    let _ = local_index.insert_chunk(chunk);
                }
                for emb in &downloaded.diff.added_embeddings {
                    let _ = local_index.insert_embedding(emb);
                }
                for edge in &downloaded.diff.added_edges {
                    let _ = local_index.add_edge(edge);
                }
                downloaded_count += 1;
            }
            Err(e) => {
                warn!(tx_id = diff_meta.tx_id, error = %e, "Failed to download diff");
            }
        }
    }

    if downloaded_count == 0 {
        return Err(crate::ValidatorError::NoContent(contributor_hex));
    }

    info!(
        contributor = hex::encode(contributor_key),
        diffs = downloaded_count,
        "Indexed contributor content"
    );

    // 3. Run test queries against the contributor's content
    let mut precision_sum = 0.0f32;
    let mut ndcg_sum = 0.0f32;
    let mut mrr_sum = 0.0f32;
    let mut query_count = 0u32;

    for query in test_queries {
        // Convert i16 query embedding to f32 for search
        let query_f32: Vec<f32> = query
            .query_embedding
            .iter()
            .map(|&v| f32::from(v) / 32767.0)
            .collect();

        let results = local_index.search(&query_f32, 10);

        match results {
            Ok(result_ids) => {
                let returned_uuids: Vec<Uuid> = result_ids.iter().map(|(id, _score)| *id).collect();

                let relevant_uuids: Vec<Uuid> = query.relevant_chunk_ids.clone();

                precision_sum += precision_at_k(&returned_uuids, &relevant_uuids, 10);
                ndcg_sum += ndcg_at_k(
                    &returned_uuids,
                    &relevant_uuids,
                    &query.relevance_grade_map(),
                    10,
                );
                mrr_sum += mrr(&returned_uuids, &relevant_uuids);
                query_count += 1;
            }
            Err(e) => {
                debug!(error = %e, "Search failed for query");
            }
        }
    }

    // 4. Verify corpus integrity: check that each document's hierarchical_hash
    // matches the Merkle root of its chunk text_hashes (Bug 7 + 8 fix)
    let mut proofs_valid = 0u32;
    let mut proofs_invalid = 0u32;
    let all_docs = local_index.get_all_documents().unwrap_or_default();
    for doc in &all_docs {
        if doc.hierarchical_hash == [0u8; 32] {
            // Placeholder hash — contributor never set it
            proofs_invalid += 1;
            continue;
        }
        let chunks = local_index.get_chunks_for_doc(doc.id).unwrap_or_default();
        if chunks.is_empty() {
            proofs_invalid += 1;
            continue;
        }
        let chunk_hashes: Vec<[u8; 32]> = chunks.iter().map(|c| c.text_hash).collect();
        let recomputed = cp_core::Document::compute_hierarchical_hash(&chunk_hashes);
        if recomputed == doc.hierarchical_hash {
            proofs_valid += 1;
        } else {
            proofs_invalid += 1;
        }
    }

    let metrics = if query_count > 0 {
        QualityMetrics {
            precision_at_10: precision_sum / query_count as f32,
            ndcg_at_10: ndcg_sum / query_count as f32,
            mrr: mrr_sum / query_count as f32,
            merkle_proofs_valid: proofs_valid,
            merkle_proofs_invalid: proofs_invalid,
            adapter_perplexity_delta: None,
        }
    } else {
        QualityMetrics {
            precision_at_10: 0.0,
            ndcg_at_10: 0.0,
            mrr: 0.0,
            merkle_proofs_valid: proofs_valid,
            merkle_proofs_invalid: proofs_invalid,
            adapter_perplexity_delta: None,
        }
    };

    let window = ValidationWindow::current();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i64;

    let mut result = EvaluationResult {
        contributor_key,
        validator_node_id,
        window_id: window.window_id,
        metrics,
        timestamp: now,
        signature: [0u8; 64],
    };

    // Sign the result
    let hash = result.signing_bytes();
    let sig = signing_key.sign(&hash);
    result.signature = sig.to_bytes();

    Ok(result)
}

/// Evaluate a local graph store directly by running test queries against it.
///
/// This extracts the query-running and metrics logic from `evaluate_contributor()`
/// into a reusable function that works against any `GraphStore` without requiring
/// Arweave download. Useful for local validation and testing.
pub fn evaluate_local_graph(
    graph: &Arc<std::sync::Mutex<GraphStore>>,
    test_queries: &[TestQuery],
    validator_node_id: [u8; 16],
    signing_key: &SigningKey,
) -> Result<EvaluationResult, crate::ValidatorError> {
    let local_index = graph
        .lock()
        .map_err(|e| crate::ValidatorError::Internal(format!("Graph lock poisoned: {e}")))?;

    let mut precision_sum = 0.0f32;
    let mut ndcg_sum = 0.0f32;
    let mut mrr_sum = 0.0f32;
    let mut query_count = 0u32;

    for query in test_queries {
        let query_f32: Vec<f32> = query
            .query_embedding
            .iter()
            .map(|&v| f32::from(v) / 32767.0)
            .collect();
        let results = local_index.search(&query_f32, 10);
        match results {
            Ok(result_ids) => {
                let returned_uuids: Vec<Uuid> = result_ids.iter().map(|(id, _score)| *id).collect();
                let relevant_uuids: Vec<Uuid> = query.relevant_chunk_ids.clone();

                precision_sum += precision_at_k(&returned_uuids, &relevant_uuids, 10);
                ndcg_sum += ndcg_at_k(
                    &returned_uuids,
                    &relevant_uuids,
                    &query.relevance_grade_map(),
                    10,
                );
                mrr_sum += mrr(&returned_uuids, &relevant_uuids);
                query_count += 1;
            }
            Err(e) => {
                debug!(error = %e, "Search failed for query");
            }
        }
    }

    // Verify corpus integrity
    let mut proofs_valid = 0u32;
    let mut proofs_invalid = 0u32;
    let all_docs = local_index.get_all_documents().unwrap_or_default();
    for doc in &all_docs {
        if doc.hierarchical_hash == [0u8; 32] {
            proofs_invalid += 1;
            continue;
        }
        let chunks = local_index.get_chunks_for_doc(doc.id).unwrap_or_default();
        if chunks.is_empty() {
            proofs_invalid += 1;
            continue;
        }
        let chunk_hashes: Vec<[u8; 32]> = chunks.iter().map(|c| c.text_hash).collect();
        let recomputed = cp_core::Document::compute_hierarchical_hash(&chunk_hashes);
        if recomputed == doc.hierarchical_hash {
            proofs_valid += 1;
        } else {
            proofs_invalid += 1;
        }
    }

    let metrics = if query_count > 0 {
        QualityMetrics {
            precision_at_10: precision_sum / query_count as f32,
            ndcg_at_10: ndcg_sum / query_count as f32,
            mrr: mrr_sum / query_count as f32,
            merkle_proofs_valid: proofs_valid,
            merkle_proofs_invalid: proofs_invalid,
            adapter_perplexity_delta: None,
        }
    } else {
        QualityMetrics {
            precision_at_10: 0.0,
            ndcg_at_10: 0.0,
            mrr: 0.0,
            merkle_proofs_valid: proofs_valid,
            merkle_proofs_invalid: proofs_invalid,
            adapter_perplexity_delta: None,
        }
    };

    let window = ValidationWindow::current();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i64;

    // Use a zero contributor_key since this is local evaluation
    let mut result = EvaluationResult {
        contributor_key: [0u8; 32],
        validator_node_id,
        window_id: window.window_id,
        metrics,
        timestamp: now,
        signature: [0u8; 64],
    };

    let hash = result.signing_bytes();
    let sig = signing_key.sign(&hash);
    result.signature = sig.to_bytes();

    Ok(result)
}

/// Evaluate a live peer by connecting over Tor, sending test queries, and
/// measuring search quality and latency.
///
/// Per CP-015 section 13. Requires a running `TorRuntime` to establish
/// connections to the peer's onion service.
pub async fn evaluate_peer(
    peer: &PeerRegistration,
    test_queries: &[TestQuery],
    validator_node_id: [u8; 16],
    signing_key: &SigningKey,
    model_hash: [u8; 32],
    tor_runtime: &cp_tor::TorRuntime,
) -> Result<PeerTestResult, crate::ValidatorError> {
    info!(
        peer = hex::encode(peer.node_id),
        onion = &peer.onion_address,
        "Testing live peer"
    );

    // Select up to 5 test queries per validation window
    let selected = if test_queries.len() > 5 {
        &test_queries[..5]
    } else {
        test_queries
    };

    let mut precision_sum = 0.0f32;
    let mut ndcg_sum = 0.0f32;
    let mut mrr_sum = 0.0f32;
    let mut latency_sum = 0u64;
    let mut valid_proofs = 0u32;
    let mut total_proofs = 0u32;
    let mut successful_queries = 0u32;

    // Derive an ephemeral session key for these queries
    let session_key = cp_tor::SessionKey::derive_with_salt(
        &signing_key.to_bytes(),
        &blake3::hash(b"validator-peer-test").as_bytes()[..32]
            .try_into()
            .unwrap(),
    );

    for query in selected {
        let start = Instant::now();

        // Connect to peer and send query
        let response_result = tokio::time::timeout(
            std::time::Duration::from_secs(5),
            send_query_to_peer(
                tor_runtime,
                &peer.onion_address,
                &query.query_embedding,
                Some(query.query_text.clone()),
                model_hash,
                &session_key,
                &peer.public_key,
            ),
        )
        .await;

        let elapsed_ms = start.elapsed().as_millis() as u64;

        match response_result {
            Ok(Ok(response)) if response.status == SearchStatus::Ok => {
                latency_sum += elapsed_ms;

                let returned_uuids: Vec<Uuid> = response
                    .results
                    .iter()
                    .map(|r| Uuid::from_bytes(r.chunk_id))
                    .collect();

                let relevant_uuids: Vec<Uuid> = query.relevant_chunk_ids.clone();

                precision_sum += precision_at_k(&returned_uuids, &relevant_uuids, 10);
                ndcg_sum += ndcg_at_k(
                    &returned_uuids,
                    &relevant_uuids,
                    &query.relevance_grade_map(),
                    10,
                );
                mrr_sum += mrr(&returned_uuids, &relevant_uuids);

                // Count proof validity
                for result in &response.results {
                    if let Some(proof_hashes) = &result.merkle_proof {
                        total_proofs += 1;
                        // The Tor search response provides proof as Vec<[u8; 32]>
                        // (sibling hashes only). To use cp_core::state::verify_merkle_proof
                        // we would need (hash, is_left) tuples and a leaf index.
                        // For now, verify the proof chain manually: hash the chunk_id
                        // as the leaf and walk up the tree. We assume left-first ordering
                        // matching the canonical Merkle tree construction.
                        let leaf = *blake3::hash(&result.chunk_id).as_bytes();
                        let computed_root = proof_hashes.iter().fold(leaf, |current, sibling| {
                            let mut hasher = blake3::Hasher::new();
                            if current <= *sibling {
                                hasher.update(&current);
                                hasher.update(sibling);
                            } else {
                                hasher.update(sibling);
                                hasher.update(&current);
                            }
                            *hasher.finalize().as_bytes()
                        });
                        if computed_root == response.peer_state_root {
                            valid_proofs += 1;
                        }
                    }
                }

                successful_queries += 1;
            }
            Ok(Ok(response)) => {
                debug!(status = ?response.status, "Peer returned non-OK status");
            }
            Ok(Err(e)) => {
                debug!(error = %e, "Search request failed");
            }
            Err(_) => {
                debug!("Search request timed out");
            }
        }
    }

    let metrics = if successful_queries > 0 {
        PeerQualityMetrics {
            precision_at_10: precision_sum / successful_queries as f32,
            ndcg_at_10: ndcg_sum / successful_queries as f32,
            mrr: mrr_sum / successful_queries as f32,
            response_latency_ms: (latency_sum / u64::from(successful_queries))
                .min(u64::from(u16::MAX)) as u16,
            proof_validity_rate: if total_proofs > 0 {
                valid_proofs as f32 / total_proofs as f32
            } else {
                0.0
            },
            availability_rate: 1.0,
        }
    } else {
        PeerQualityMetrics {
            precision_at_10: 0.0,
            ndcg_at_10: 0.0,
            mrr: 0.0,
            response_latency_ms: 0,
            proof_validity_rate: 0.0,
            availability_rate: 0.0,
        }
    };

    let window = ValidationWindow::current();
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i64;

    let mut result = PeerTestResult {
        peer_node_id: peer.node_id,
        peer_onion: peer.onion_address.clone(),
        validator_node_id,
        window_id: window.window_id,
        metrics,
        timestamp: now,
        signature: [0u8; 64],
    };

    let hash = result.signing_bytes();
    let sig = signing_key.sign(&hash);
    result.signature = sig.to_bytes();

    Ok(result)
}

/// Send a search query to a peer over Tor using the cp-tor client.
async fn send_query_to_peer(
    tor_runtime: &cp_tor::TorRuntime,
    onion_address: &str,
    query_embedding: &[i16],
    query_text: Option<String>,
    model_hash: [u8; 32],
    session_key: &cp_tor::SessionKey,
    peer_public_key: &[u8; 32],
) -> Result<SearchResponse, crate::ValidatorError> {
    let mut stream = tor_runtime
        .connect_to_peer(onion_address)
        .await
        .map_err(|e| crate::ValidatorError::Tor(e.to_string()))?;

    let response = cp_tor::search(
        &mut stream,
        query_embedding.to_vec(),
        query_text,
        10,
        true,
        model_hash,
        session_key,
        peer_public_key,
    )
    .await
    .map_err(|e| crate::ValidatorError::Tor(e.to_string()))?;

    Ok(response)
}

/// Update contributor ratings based on evaluation results using pairwise
/// `OpenSkill` comparisons.
///
/// Per CP-015 section 7: Results are sorted by composite score, then
/// adjacent pairs are compared using the `OpenSkill` update rule.
pub fn update_contributor_ratings(
    results: &[EvaluationResult],
    ratings: &mut HashMap<[u8; 32], (Rating, u64, i64)>,
) {
    if results.len() < 2 {
        return;
    }

    // Sort by composite score, descending
    let mut ranked: Vec<&EvaluationResult> = results.iter().collect();
    ranked.sort_by(|a, b| {
        composite_score(&b.metrics)
            .partial_cmp(&composite_score(&a.metrics))
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i64;

    // All-pairs comparisons: every higher-ranked entry beats every lower-ranked one
    for i in 0..ranked.len() {
        for j in (i + 1)..ranked.len() {
            let winner_key = ranked[i].contributor_key;
            let loser_key = ranked[j].contributor_key;

            let winner_entry = ratings
                .entry(winner_key)
                .or_insert((Rating::default(), 0, now));
            let winner_rating = winner_entry.0;

            let loser_entry = ratings
                .entry(loser_key)
                .or_insert((Rating::default(), 0, now));
            let loser_rating = loser_entry.0;

            let (new_winner, new_loser) =
                crate::rating::pairwise_update(&winner_rating, &loser_rating);

            ratings.get_mut(&winner_key).unwrap().0 = new_winner;
            ratings.get_mut(&winner_key).unwrap().1 += 1;
            ratings.get_mut(&winner_key).unwrap().2 = now;

            ratings.get_mut(&loser_key).unwrap().0 = new_loser;
            ratings.get_mut(&loser_key).unwrap().1 += 1;
            ratings.get_mut(&loser_key).unwrap().2 = now;
        }
    }
}

/// Update peer ratings based on peer test results using pairwise
/// `OpenSkill` comparisons.
///
/// Same mechanism as contributor ratings (CP-015 section 14) but uses
/// `peer_composite_score` for ranking.
pub fn update_peer_ratings(
    results: &[PeerTestResult],
    ratings: &mut HashMap<[u8; 16], (Rating, u64, i64)>,
) {
    if results.len() < 2 {
        return;
    }

    let mut ranked: Vec<&PeerTestResult> = results.iter().collect();
    ranked.sort_by(|a, b| {
        peer_composite_score(&b.metrics)
            .partial_cmp(&peer_composite_score(&a.metrics))
            .unwrap_or(std::cmp::Ordering::Equal)
    });

    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap()
        .as_millis() as i64;

    // All-pairs comparisons: every higher-ranked entry beats every lower-ranked one
    for i in 0..ranked.len() {
        for j in (i + 1)..ranked.len() {
            let winner_key = ranked[i].peer_node_id;
            let loser_key = ranked[j].peer_node_id;

            let winner_entry = ratings
                .entry(winner_key)
                .or_insert((Rating::default(), 0, now));
            let winner_rating = winner_entry.0;

            let loser_entry = ratings
                .entry(loser_key)
                .or_insert((Rating::default(), 0, now));
            let loser_rating = loser_entry.0;

            let (new_winner, new_loser) =
                crate::rating::pairwise_update(&winner_rating, &loser_rating);

            ratings.get_mut(&winner_key).unwrap().0 = new_winner;
            ratings.get_mut(&winner_key).unwrap().1 += 1;
            ratings.get_mut(&winner_key).unwrap().2 = now;

            ratings.get_mut(&loser_key).unwrap().0 = new_loser;
            ratings.get_mut(&loser_key).unwrap().1 += 1;
            ratings.get_mut(&loser_key).unwrap().2 = now;
        }
    }
}

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

    fn make_uuids(n: usize) -> Vec<Uuid> {
        (0..n)
            .map(|i| {
                let mut bytes = [0u8; 16];
                bytes[0] = i as u8;
                Uuid::from_bytes(bytes)
            })
            .collect()
    }

    // ---- precision_at_k ----

    #[test]
    fn test_precision_at_k_all_relevant() {
        let ids = make_uuids(10);
        let relevant = ids.clone();
        assert!((precision_at_k(&ids, &relevant, 10) - 1.0).abs() < 1e-6);
    }

    #[test]
    fn test_precision_at_k_none_relevant() {
        let returned = make_uuids(10);
        let relevant = vec![Uuid::from_bytes([99u8; 16])];
        assert!((precision_at_k(&returned, &relevant, 10)).abs() < 1e-6);
    }

    #[test]
    fn test_precision_at_k_half_relevant() {
        let returned = make_uuids(10);
        let relevant: Vec<Uuid> = returned[..5].to_vec();
        assert!((precision_at_k(&returned, &relevant, 10) - 0.5).abs() < 1e-6);
    }

    #[test]
    fn test_precision_at_k_fewer_results_than_k() {
        let returned = make_uuids(3);
        let relevant = returned.clone();
        // 3 relevant out of k=10 -> 3/10 = 0.3
        assert!((precision_at_k(&returned, &relevant, 10) - 0.3).abs() < 1e-6);
    }

    #[test]
    fn test_precision_at_k_zero() {
        let returned = make_uuids(5);
        let relevant = returned.clone();
        assert!((precision_at_k(&returned, &relevant, 0)).abs() < 1e-6);
    }

    #[test]
    fn test_precision_at_k_empty_returned() {
        let relevant = make_uuids(5);
        assert!((precision_at_k(&[], &relevant, 10)).abs() < 1e-6);
    }

    // ---- ndcg_at_k ----

    #[test]
    fn test_ndcg_at_k_perfect_ranking() {
        let ids = make_uuids(5);
        let relevant = ids.clone();
        let mut grades = HashMap::new();
        for (i, id) in ids.iter().enumerate() {
            grades.insert(*id, (5 - i) as u8);
        }
        let score = ndcg_at_k(&ids, &relevant, &grades, 5);
        assert!(
            (score - 1.0).abs() < 1e-5,
            "Perfect NDCG should be 1.0, got {score}"
        );
    }

    #[test]
    fn test_ndcg_at_k_no_relevant() {
        let returned = make_uuids(5);
        let relevant: Vec<Uuid> = vec![];
        let grades = HashMap::new();
        assert!((ndcg_at_k(&returned, &relevant, &grades, 5)).abs() < 1e-6);
    }

    #[test]
    fn test_ndcg_at_k_binary_relevance() {
        let returned = make_uuids(5);
        let relevant = vec![returned[4]];
        let grades = HashMap::new();

        let score = ndcg_at_k(&returned, &relevant, &grades, 5);
        assert!(score > 0.3 && score < 0.5, "Got NDCG = {score}");
    }

    #[test]
    fn test_ndcg_at_k_zero_k() {
        let ids = make_uuids(5);
        let grades = HashMap::new();
        assert!((ndcg_at_k(&ids, &ids, &grades, 0)).abs() < 1e-6);
    }

    // ---- mrr ----

    #[test]
    fn test_mrr_first_position() {
        let returned = make_uuids(5);
        let relevant = vec![returned[0]];
        assert!((mrr(&returned, &relevant) - 1.0).abs() < 1e-6);
    }

    #[test]
    fn test_mrr_second_position() {
        let returned = make_uuids(5);
        let relevant = vec![returned[1]];
        assert!((mrr(&returned, &relevant) - 0.5).abs() < 1e-6);
    }

    #[test]
    fn test_mrr_third_position() {
        let returned = make_uuids(5);
        let relevant = vec![returned[2]];
        let score = mrr(&returned, &relevant);
        assert!(
            (score - 1.0 / 3.0).abs() < 1e-6,
            "Expected 1/3, got {score}"
        );
    }

    #[test]
    fn test_mrr_no_relevant() {
        let returned = make_uuids(5);
        let relevant = vec![Uuid::from_bytes([99u8; 16])];
        assert!((mrr(&returned, &relevant)).abs() < 1e-6);
    }

    #[test]
    fn test_mrr_empty() {
        let relevant = make_uuids(3);
        assert!((mrr(&[], &relevant)).abs() < 1e-6);
    }

    #[test]
    fn test_mrr_multiple_relevant_returns_first() {
        let returned = make_uuids(5);
        let relevant = vec![returned[1], returned[3]];
        assert!((mrr(&returned, &relevant) - 0.5).abs() < 1e-6);
    }

    // ---- composite_score ----

    #[test]
    fn test_composite_score_perfect() {
        let metrics = QualityMetrics {
            precision_at_10: 1.0,
            ndcg_at_10: 1.0,
            mrr: 1.0,
            merkle_proofs_valid: 10,
            merkle_proofs_invalid: 0,
            adapter_perplexity_delta: None,
        };
        let score = composite_score(&metrics);
        assert!(
            (score - 1.0).abs() < 1e-6,
            "Perfect score should be 1.0, got {score}"
        );
    }

    #[test]
    fn test_composite_score_zero() {
        let metrics = QualityMetrics {
            precision_at_10: 0.0,
            ndcg_at_10: 0.0,
            mrr: 0.0,
            merkle_proofs_valid: 0,
            merkle_proofs_invalid: 0,
            adapter_perplexity_delta: None,
        };
        assert!((composite_score(&metrics)).abs() < 1e-6);
    }

    #[test]
    fn test_composite_score_adapter_penalty() {
        let metrics = QualityMetrics {
            precision_at_10: 1.0,
            ndcg_at_10: 1.0,
            mrr: 1.0,
            merkle_proofs_valid: 10,
            merkle_proofs_invalid: 0,
            adapter_perplexity_delta: Some(0.5),
        };
        let score = composite_score(&metrics);
        assert!(
            (score - 0.8).abs() < 1e-6,
            "Penalized score should be 0.8, got {score}"
        );
    }

    #[test]
    fn test_composite_score_adapter_improvement() {
        let metrics = QualityMetrics {
            precision_at_10: 1.0,
            ndcg_at_10: 1.0,
            mrr: 1.0,
            merkle_proofs_valid: 10,
            merkle_proofs_invalid: 0,
            adapter_perplexity_delta: Some(-0.5),
        };
        let score = composite_score(&metrics);
        assert!((score - 1.0).abs() < 1e-6);
    }

    // ---- peer_composite_score ----

    #[test]
    fn test_peer_composite_score_fast_good_peer() {
        let metrics = PeerQualityMetrics {
            precision_at_10: 1.0,
            ndcg_at_10: 1.0,
            mrr: 1.0,
            response_latency_ms: 0,
            proof_validity_rate: 1.0,
            availability_rate: 1.0,
        };
        let score = peer_composite_score(&metrics);
        // quality = 0.4 + 0.2 + 0.1 = 0.7
        // reliability = 1.0
        // latency_score = 1.0
        // 0.7*0.7 + 0.2*1.0 + 0.1*1.0 = 0.49 + 0.2 + 0.1 = 0.79
        assert!((score - 0.79).abs() < 0.01, "Got {score}");
    }

    #[test]
    fn test_peer_composite_score_slow_peer() {
        let metrics = PeerQualityMetrics {
            precision_at_10: 1.0,
            ndcg_at_10: 1.0,
            mrr: 1.0,
            response_latency_ms: 5000,
            proof_validity_rate: 1.0,
            availability_rate: 1.0,
        };
        let score = peer_composite_score(&metrics);
        assert!((score - 0.69).abs() < 0.01, "Got {score}");
    }

    #[test]
    fn test_peer_composite_score_unreachable() {
        let metrics = PeerQualityMetrics {
            precision_at_10: 0.0,
            ndcg_at_10: 0.0,
            mrr: 0.0,
            response_latency_ms: 0,
            proof_validity_rate: 0.0,
            availability_rate: 0.0,
        };
        let score = peer_composite_score(&metrics);
        assert!((score - 0.1).abs() < 0.01, "Got {score}");
    }

    // ---- update_contributor_ratings ----

    #[test]
    fn test_update_contributor_ratings_single_result() {
        let result = EvaluationResult {
            contributor_key: [1u8; 32],
            validator_node_id: [0u8; 16],
            window_id: 100,
            metrics: QualityMetrics {
                precision_at_10: 0.8,
                ndcg_at_10: 0.7,
                mrr: 0.9,
                merkle_proofs_valid: 5,
                merkle_proofs_invalid: 0,
                adapter_perplexity_delta: None,
            },
            timestamp: 0,
            signature: [0u8; 64],
        };

        let mut ratings = HashMap::new();
        update_contributor_ratings(&[result], &mut ratings);
        assert!(ratings.is_empty());
    }

    #[test]
    fn test_update_contributor_ratings_two_results() {
        let good = EvaluationResult {
            contributor_key: [1u8; 32],
            validator_node_id: [0u8; 16],
            window_id: 100,
            metrics: QualityMetrics {
                precision_at_10: 0.9,
                ndcg_at_10: 0.9,
                mrr: 0.9,
                merkle_proofs_valid: 10,
                merkle_proofs_invalid: 0,
                adapter_perplexity_delta: None,
            },
            timestamp: 0,
            signature: [0u8; 64],
        };

        let bad = EvaluationResult {
            contributor_key: [2u8; 32],
            validator_node_id: [0u8; 16],
            window_id: 100,
            metrics: QualityMetrics {
                precision_at_10: 0.1,
                ndcg_at_10: 0.1,
                mrr: 0.1,
                merkle_proofs_valid: 1,
                merkle_proofs_invalid: 9,
                adapter_perplexity_delta: None,
            },
            timestamp: 0,
            signature: [0u8; 64],
        };

        let mut ratings = HashMap::new();
        update_contributor_ratings(&[good, bad], &mut ratings);

        let good_rating = ratings.get(&[1u8; 32]).unwrap();
        let bad_rating = ratings.get(&[2u8; 32]).unwrap();

        assert!(
            good_rating.0.mu > bad_rating.0.mu,
            "Good contributor mu ({}) should exceed bad contributor mu ({})",
            good_rating.0.mu,
            bad_rating.0.mu
        );
    }
}