cp_validator/

lib.rs

//! Validator protocol for empirical quality verification (CP-015).
//!
//! Any Canon node can opt into validation. Validators download content from
//! Arweave, index it locally, run test queries against it, measure retrieval
//! quality, and publish `OpenSkill` Bayesian ratings for content contributors.
//! These ratings help other nodes prioritize what to download from Arweave.
//!
//! Validators also test live search peers by connecting over Tor, sending
//! test queries, and rating peer quality for circuit selection.
//!
//! The approach adapts Templar/Covenant (tplr.ai), which uses empirical
//! quality verification for decentralized pre-training. `OpenSkill` Bayesian
//! ratings provide a robust, game-resistant ranking that converges even
//! with sparse observations. Window-based coordination ensures validators
//! test the same content without requiring blockchain synchronization.
//!
//! Data flow:
//!   1. Load test query corpus (from Arweave or local file)
//!   2. Compute current validation window
//!   3. Select test queries deterministically from corpus
//!   4. For each contributor: download from Arweave, index, run queries, measure metrics
//!   5. Rank contributors by composite score, update `OpenSkill` ratings pairwise
//!   6. For each peer: connect over Tor, send queries, measure quality + latency
//!   7. Rank peers by composite score, update `OpenSkill` ratings pairwise
//!   8. Publish results to Arweave

pub mod corpus;
pub mod evaluation;
pub mod rating;
pub mod window;

pub use corpus::{TestCorpus, TestQuery};
pub use evaluation::{
    composite_score, evaluate_contributor, evaluate_local_graph, evaluate_peer, mrr, ndcg_at_k,
    peer_composite_score, precision_at_k, update_contributor_ratings, update_peer_ratings,
    EvaluationResult, PeerQualityMetrics, PeerTestResult, QualityMetrics,
};
pub use rating::{pairwise_update, Rating, DEFAULT_MU, DEFAULT_SIGMA};
pub use window::{select_test_queries, ValidationWindow};

use cp_arweave::ArweaveClient;
use ed25519_dalek::SigningKey;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, info, warn};

/// Errors specific to the validator protocol.
#[derive(Debug, thiserror::Error)]
pub enum ValidatorError {
    #[error("Arweave error: {0}")]
    Arweave(String),

    #[error("No content found for contributor: {0}")]
    NoContent(String),

    #[error("Corpus error: {0}")]
    Corpus(String),

    #[error("Tor not available")]
    TorNotAvailable,

    #[error("Tor error: {0}")]
    Tor(String),

    #[error("Internal error: {0}")]
    Internal(String),
}

/// Configuration for the validator.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidatorConfig {
    /// How often to run the evaluation loop (seconds).
    /// Defaults to 3600 (once per validation window).
    pub evaluation_interval_secs: u64,

    /// Path to a local test corpus file. If set, this is used instead of
    /// downloading from Arweave. Useful for bootstrapping and testing.
    pub corpus_path: Option<PathBuf>,

    /// Number of test queries to select per evaluation window.
    pub queries_per_window: usize,

    /// Maximum number of contributors to evaluate per window.
    pub max_contributors_per_window: usize,

    /// Maximum number of peers to test per window.
    pub max_peers_per_window: usize,

    /// Whether to test live peers over Tor.
    pub test_peers: bool,
}

impl Default for ValidatorConfig {
    fn default() -> Self {
        Self {
            evaluation_interval_secs: 3600,
            corpus_path: None,
            queries_per_window: 20,
            max_contributors_per_window: 50,
            max_peers_per_window: 20,
            test_peers: true,
        }
    }
}

/// `OpenSkill` Bayesian rating for a content contributor.
///
/// Per CP-015 section 2.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContributorRating {
    /// Ed25519 public key of the contributor.
    pub contributor_key: [u8; 32],
    /// `OpenSkill` mean skill estimate.
    pub mu: f64,
    /// `OpenSkill` uncertainty.
    pub sigma: f64,
    /// Number of times this contributor has been validated.
    pub validation_count: u64,
    /// Last updated timestamp (Unix milliseconds).
    pub last_updated: i64,
}

impl ContributorRating {
    /// Create a new contributor rating with default `OpenSkill` values.
    pub fn new(contributor_key: [u8; 32]) -> Self {
        Self {
            contributor_key,
            mu: DEFAULT_MU,
            sigma: DEFAULT_SIGMA,
            validation_count: 0,
            last_updated: 0,
        }
    }

    /// Conservative score: mu - 2*sigma.
    pub fn conservative_score(&self) -> f64 {
        self.mu - 2.0 * self.sigma
    }

    /// Convert to a Rating struct for the `OpenSkill` update functions.
    pub fn to_rating(&self) -> Rating {
        Rating::new(self.mu, self.sigma)
    }

    /// Update from a Rating struct after an `OpenSkill` update.
    pub fn apply_rating(&mut self, rating: &Rating, now_ms: i64) {
        self.mu = rating.mu;
        self.sigma = rating.sigma;
        self.validation_count += 1;
        self.last_updated = now_ms;
    }
}

/// `OpenSkill` Bayesian rating for a live search peer.
///
/// Per CP-015 section 14. Same `OpenSkill` model as `ContributorRating` but
/// tracked independently. A contributor's content quality (measured by
/// downloading and indexing) is distinct from their peer quality
/// (measured by connecting and querying live).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PeerRating {
    /// Node ID of the peer.
    pub peer_node_id: [u8; 16],
    /// `OpenSkill` mean skill estimate.
    pub mu: f64,
    /// `OpenSkill` uncertainty.
    pub sigma: f64,
    /// Number of peer tests.
    pub test_count: u64,
    /// Last updated timestamp (Unix milliseconds).
    pub last_updated: i64,
}

impl PeerRating {
    /// Create a new peer rating with default `OpenSkill` values.
    pub fn new(peer_node_id: [u8; 16]) -> Self {
        Self {
            peer_node_id,
            mu: DEFAULT_MU,
            sigma: DEFAULT_SIGMA,
            test_count: 0,
            last_updated: 0,
        }
    }

    /// Conservative score: mu - 2*sigma.
    pub fn conservative_score(&self) -> f64 {
        self.mu - 2.0 * self.sigma
    }

    /// Convert to a Rating struct.
    pub fn to_rating(&self) -> Rating {
        Rating::new(self.mu, self.sigma)
    }

    /// Update from a Rating struct.
    pub fn apply_rating(&mut self, rating: &Rating, now_ms: i64) {
        self.mu = rating.mu;
        self.sigma = rating.sigma;
        self.test_count += 1;
        self.last_updated = now_ms;
    }
}

/// The main validator that runs evaluation loops.
///
/// Orchestrates the full validation pipeline: loading the test corpus,
/// selecting queries, evaluating contributors, updating ratings.
/// Peer testing requires a `TorRuntime` to be passed in.
pub struct Validator {
    /// Validator configuration.
    config: ValidatorConfig,

    /// Arweave client for content download.
    arweave: Arc<ArweaveClient>,

    /// Ed25519 signing key for this validator node.
    signing_key: SigningKey,

    /// Node ID derived from the signing key.
    node_id: [u8; 16],

    /// Contributor ratings indexed by public key.
    contributor_ratings: Arc<RwLock<HashMap<[u8; 32], ContributorRating>>>,

    /// Peer ratings indexed by node ID.
    peer_ratings: Arc<RwLock<HashMap<[u8; 16], PeerRating>>>,

    /// The test query corpus.
    corpus: Arc<RwLock<Option<TestCorpus>>>,

    /// BLAKE3 hash of the embedding model, for compatibility checking.
    model_hash: [u8; 32],
}

impl Validator {
    /// Create a new validator.
    pub fn new(
        config: ValidatorConfig,
        arweave: Arc<ArweaveClient>,
        signing_key: SigningKey,
        model_hash: [u8; 32],
    ) -> Self {
        let public_key = signing_key.verifying_key().to_bytes();
        let node_id_hash = blake3::hash(&public_key);
        let mut node_id = [0u8; 16];
        node_id.copy_from_slice(&node_id_hash.as_bytes()[..16]);

        Self {
            config,
            arweave,
            signing_key,
            node_id,
            contributor_ratings: Arc::new(RwLock::new(HashMap::new())),
            peer_ratings: Arc::new(RwLock::new(HashMap::new())),
            corpus: Arc::new(RwLock::new(None)),
            model_hash,
        }
    }

    /// Get a read reference to the contributor ratings.
    pub fn contributor_ratings(&self) -> &Arc<RwLock<HashMap<[u8; 32], ContributorRating>>> {
        &self.contributor_ratings
    }

    /// Get a read reference to the peer ratings.
    pub fn peer_ratings(&self) -> &Arc<RwLock<HashMap<[u8; 16], PeerRating>>> {
        &self.peer_ratings
    }

    /// Get this validator's node ID.
    pub fn node_id(&self) -> [u8; 16] {
        self.node_id
    }

    /// Load or refresh the test corpus.
    ///
    /// Uses the local file if configured, otherwise downloads from Arweave.
    pub async fn load_corpus(&self) -> Result<(), ValidatorError> {
        let new_corpus = if let Some(ref path) = self.config.corpus_path {
            TestCorpus::load_from_file(path)?
        } else {
            TestCorpus::load_from_arweave(&self.arweave).await?
        };

        info!(queries = new_corpus.len(), "Loaded test corpus");
        *self.corpus.write().await = Some(new_corpus);
        Ok(())
    }

    /// Run a single evaluation cycle for the current validation window.
    ///
    /// This is the core validation loop:
    /// 1. Ensure corpus is loaded
    /// 2. Select test queries for the current window
    /// 3. Discover contributors on Arweave
    /// 4. Evaluate each contributor
    /// 5. Update contributor ratings
    ///
    /// Returns the evaluation results for optional publication to Arweave.
    pub async fn run_evaluation_cycle(&self) -> Result<Vec<EvaluationResult>, ValidatorError> {
        // 1. Ensure corpus is loaded
        {
            let corpus_guard = self.corpus.read().await;
            if corpus_guard.is_none() {
                drop(corpus_guard);
                self.load_corpus().await?;
            }
        }

        let corpus = self.corpus.read().await;
        let corpus = corpus
            .as_ref()
            .ok_or_else(|| ValidatorError::Corpus("Corpus not loaded".to_string()))?;

        // 2. Select test queries for current window
        let window = ValidationWindow::current();
        let test_queries = window.select_test_queries(corpus, self.config.queries_per_window);

        if test_queries.is_empty() {
            warn!("No test queries selected for window {}", window.window_id);
            return Ok(Vec::new());
        }

        info!(
            window_id = window.window_id,
            queries = test_queries.len(),
            "Starting evaluation cycle"
        );

        // 3. Discover contributors on Arweave
        let (discovered, _cursor) = cp_arweave::discover_diffs(&self.arweave, None, 100, None)
            .await
            .map_err(|e| ValidatorError::Arweave(e.to_string()))?;

        // Extract unique contributor keys
        let mut contributor_keys: Vec<[u8; 32]> = Vec::new();
        for diff in &discovered {
            if let Ok(key_bytes) = hex::decode(&diff.contributor_key) {
                if key_bytes.len() == 32 {
                    let mut key = [0u8; 32];
                    key.copy_from_slice(&key_bytes);
                    if !contributor_keys.contains(&key) {
                        contributor_keys.push(key);
                    }
                }
            }
        }

        contributor_keys.truncate(self.config.max_contributors_per_window);

        info!(
            contributors = contributor_keys.len(),
            "Discovered contributors to evaluate"
        );

        // 4. Evaluate each contributor
        let mut eval_results = Vec::new();
        for contributor_key in &contributor_keys {
            match evaluate_contributor(
                *contributor_key,
                &self.arweave,
                &test_queries,
                self.node_id,
                &self.signing_key,
            )
            .await
            {
                Ok(result) => {
                    eval_results.push(result);
                }
                Err(e) => {
                    warn!(
                        contributor = hex::encode(contributor_key),
                        error = %e,
                        "Failed to evaluate contributor"
                    );
                }
            }
        }

        // 5. Update contributor ratings
        if eval_results.len() >= 2 {
            let mut raw_ratings: HashMap<[u8; 32], (Rating, u64, i64)> = HashMap::new();

            {
                let existing = self.contributor_ratings.read().await;
                for (key, cr) in existing.iter() {
                    raw_ratings
                        .insert(*key, (cr.to_rating(), cr.validation_count, cr.last_updated));
                }
            }

            update_contributor_ratings(&eval_results, &mut raw_ratings);

            {
                let mut ratings = self.contributor_ratings.write().await;
                for (key, (r, count, updated)) in &raw_ratings {
                    let entry = ratings
                        .entry(*key)
                        .or_insert_with(|| ContributorRating::new(*key));
                    entry.mu = r.mu;
                    entry.sigma = r.sigma;
                    entry.validation_count = *count;
                    entry.last_updated = *updated;
                }
            }

            info!(
                evaluated = eval_results.len(),
                "Updated contributor ratings"
            );
        }

        Ok(eval_results)
    }

    /// Test live peers by connecting over Tor and running test queries.
    ///
    /// Requires a running `TorRuntime`. Returns the peer test results
    /// for optional publication to Arweave.
    pub async fn test_live_peers(
        &self,
        tor_runtime: &cp_tor::TorRuntime,
    ) -> Result<Vec<PeerTestResult>, ValidatorError> {
        // Ensure corpus is loaded
        let corpus = self.corpus.read().await;
        let corpus = corpus
            .as_ref()
            .ok_or_else(|| ValidatorError::Corpus("Corpus not loaded".to_string()))?;

        let window = ValidationWindow::current();
        let test_queries = window.select_test_queries(corpus, self.config.queries_per_window);

        // Discover peers on Arweave
        let model_hash_hex = hex::encode(self.model_hash);
        let (peers, _cursor) =
            cp_arweave::discover_peers(&self.arweave, Some(&model_hash_hex), 50, None)
                .await
                .map_err(|e| ValidatorError::Arweave(e.to_string()))?;

        let peers_to_test: Vec<_> = peers
            .into_iter()
            .take(self.config.max_peers_per_window)
            .collect();

        info!(peers = peers_to_test.len(), "Testing live peers");

        let mut peer_results = Vec::new();
        for peer_meta in &peers_to_test {
            match cp_arweave::download_peer_registration(&self.arweave, peer_meta).await {
                Ok(registration) => {
                    match evaluate_peer(
                        &registration,
                        &test_queries,
                        self.node_id,
                        &self.signing_key,
                        self.model_hash,
                        tor_runtime,
                    )
                    .await
                    {
                        Ok(result) => {
                            peer_results.push(result);
                        }
                        Err(e) => {
                            warn!(
                                peer = hex::encode(registration.node_id),
                                error = %e,
                                "Failed to test peer"
                            );
                        }
                    }
                }
                Err(e) => {
                    warn!(tx_id = &peer_meta.tx_id, error = %e, "Failed to download peer registration");
                }
            }
        }

        // Update peer ratings
        if peer_results.len() >= 2 {
            let mut raw_ratings: HashMap<[u8; 16], (Rating, u64, i64)> = HashMap::new();

            {
                let existing = self.peer_ratings.read().await;
                for (key, pr) in existing.iter() {
                    raw_ratings.insert(*key, (pr.to_rating(), pr.test_count, pr.last_updated));
                }
            }

            update_peer_ratings(&peer_results, &mut raw_ratings);

            {
                let mut ratings = self.peer_ratings.write().await;
                for (key, (r, count, updated)) in &raw_ratings {
                    let entry = ratings.entry(*key).or_insert_with(|| PeerRating::new(*key));
                    entry.mu = r.mu;
                    entry.sigma = r.sigma;
                    entry.test_count = *count;
                    entry.last_updated = *updated;
                }
            }

            info!(tested = peer_results.len(), "Updated peer ratings");
        }

        Ok(peer_results)
    }

    /// Run the validator loop continuously.
    ///
    /// Runs evaluation cycles at the configured interval, sleeping
    /// between cycles. If peer testing is enabled and a `TorRuntime` is
    /// provided, also tests live peers each cycle.
    pub async fn run_loop(
        &self,
        tor_runtime: Option<&cp_tor::TorRuntime>,
    ) -> Result<(), ValidatorError> {
        info!(
            interval_secs = self.config.evaluation_interval_secs,
            test_peers = self.config.test_peers,
            "Starting validator loop"
        );

        loop {
            // Contributor evaluation
            match self.run_evaluation_cycle().await {
                Ok(results) => {
                    info!(results = results.len(), "Evaluation cycle completed");
                }
                Err(e) => {
                    error!(error = %e, "Evaluation cycle failed");
                }
            }

            // Live peer testing (if enabled and Tor is available)
            if self.config.test_peers {
                if let Some(runtime) = tor_runtime {
                    match self.test_live_peers(runtime).await {
                        Ok(results) => {
                            info!(peers_tested = results.len(), "Peer testing completed");
                        }
                        Err(e) => {
                            warn!(error = %e, "Peer testing failed");
                        }
                    }
                } else {
                    warn!("Peer testing enabled but no TorRuntime provided");
                }
            }

            tokio::time::sleep(std::time::Duration::from_secs(
                self.config.evaluation_interval_secs,
            ))
            .await;
        }
    }
}

/// Prioritize Arweave downloads by contributor rating.
///
/// Per CP-015 section 9: content from well-rated contributors is
/// downloaded first. Uses conservative score (mu - 2*sigma) so
/// uncertain contributors are ranked lower than confidently-good ones.
pub fn prioritize_downloads(
    available_tx_ids: &[(String, Option<[u8; 32]>)],
    contributor_ratings: &HashMap<[u8; 32], ContributorRating>,
) -> Vec<String> {
    let mut scored: Vec<(&str, f64)> = available_tx_ids
        .iter()
        .map(|(tx_id, contributor_key)| {
            let score = contributor_key
                .and_then(|key| contributor_ratings.get(&key))
                .map_or(
                    DEFAULT_MU - 2.0 * DEFAULT_SIGMA,
                    ContributorRating::conservative_score,
                );
            (tx_id.as_str(), score)
        })
        .collect();

    scored.sort_by(|(_, a), (_, b)| b.partial_cmp(a).unwrap_or(std::cmp::Ordering::Equal));

    scored
        .into_iter()
        .map(|(tx_id, _)| tx_id.to_string())
        .collect()
}

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

    #[test]
    fn test_contributor_rating_default() {
        let cr = ContributorRating::new([1u8; 32]);
        assert!((cr.mu - 25.0).abs() < 1e-6);
        assert!((cr.sigma - DEFAULT_SIGMA).abs() < 1e-6);
        assert_eq!(cr.validation_count, 0);
    }

    #[test]
    fn test_contributor_rating_conservative_score() {
        let cr = ContributorRating {
            contributor_key: [1u8; 32],
            mu: 30.0,
            sigma: 5.0,
            validation_count: 10,
            last_updated: 0,
        };
        assert!((cr.conservative_score() - 20.0).abs() < 1e-6);
    }

    #[test]
    fn test_contributor_rating_apply() {
        let mut cr = ContributorRating::new([1u8; 32]);
        let new_rating = Rating::new(28.0, 6.5);
        cr.apply_rating(&new_rating, 1000);

        assert!((cr.mu - 28.0).abs() < 1e-6);
        assert!((cr.sigma - 6.5).abs() < 1e-6);
        assert_eq!(cr.validation_count, 1);
        assert_eq!(cr.last_updated, 1000);
    }

    #[test]
    fn test_peer_rating_default() {
        let pr = PeerRating::new([1u8; 16]);
        assert!((pr.mu - 25.0).abs() < 1e-6);
        assert!((pr.sigma - DEFAULT_SIGMA).abs() < 1e-6);
        assert_eq!(pr.test_count, 0);
    }

    #[test]
    fn test_peer_rating_apply() {
        let mut pr = PeerRating::new([1u8; 16]);
        let new_rating = Rating::new(27.0, 7.0);
        pr.apply_rating(&new_rating, 2000);

        assert!((pr.mu - 27.0).abs() < 1e-6);
        assert!((pr.sigma - 7.0).abs() < 1e-6);
        assert_eq!(pr.test_count, 1);
        assert_eq!(pr.last_updated, 2000);
    }

    #[test]
    fn test_validator_config_default() {
        let config = ValidatorConfig::default();
        assert_eq!(config.evaluation_interval_secs, 3600);
        assert_eq!(config.queries_per_window, 20);
        assert_eq!(config.max_contributors_per_window, 50);
        assert_eq!(config.max_peers_per_window, 20);
        assert!(config.test_peers);
    }

    #[test]
    fn test_prioritize_downloads_rated() {
        let mut ratings = HashMap::new();

        ratings.insert(
            [1u8; 32],
            ContributorRating {
                contributor_key: [1u8; 32],
                mu: 35.0,
                sigma: 3.0,
                validation_count: 20,
                last_updated: 0,
            },
        );

        ratings.insert(
            [2u8; 32],
            ContributorRating {
                contributor_key: [2u8; 32],
                mu: 15.0,
                sigma: 3.0,
                validation_count: 20,
                last_updated: 0,
            },
        );

        let txs = vec![
            ("tx_bad".to_string(), Some([2u8; 32])),
            ("tx_good".to_string(), Some([1u8; 32])),
            ("tx_unknown".to_string(), None),
        ];

        let prioritized = prioritize_downloads(&txs, &ratings);

        assert_eq!(prioritized[0], "tx_good");
        // Bad contributor has conservative_score = 15.0 - 6.0 = 9.0
        // Unknown contributor has default = 25.0 - 16.666 = 8.333
        // So bad comes before unknown
        assert_eq!(prioritized[1], "tx_bad");
        assert_eq!(prioritized[2], "tx_unknown");
    }

    #[test]
    fn test_prioritize_downloads_empty() {
        let ratings = HashMap::new();
        let txs: Vec<(String, Option<[u8; 32]>)> = Vec::new();
        let prioritized = prioritize_downloads(&txs, &ratings);
        assert!(prioritized.is_empty());
    }

    #[test]
    fn test_prioritize_downloads_all_unrated() {
        let ratings = HashMap::new();
        let txs = vec![
            ("tx1".to_string(), Some([1u8; 32])),
            ("tx2".to_string(), Some([2u8; 32])),
        ];

        let prioritized = prioritize_downloads(&txs, &ratings);
        assert_eq!(prioritized.len(), 2);
    }
}