cp_validator/

window.rs

//! Window-based coordination for validators.
//!
//! Per CP-015 section 3: Validators coordinate through deterministic time
//! windows without blockchain synchronization. All validators in the same
//! hourly window independently select the same test queries from the
//! canonical corpus using a deterministic RNG seeded by BLAKE3(window_id).

use crate::corpus::{TestCorpus, TestQuery};
use rand::seq::SliceRandom;
use rand::SeedableRng;
use rand_chacha::ChaCha20Rng;
use serde::{Deserialize, Serialize};

/// Duration of each validation window in seconds.
const WINDOW_DURATION_SECS: u64 = 3600; // 1 hour

/// A deterministic validation window derived from the current time.
///
/// All validators observing the same wall clock hour will compute
/// the same window_id and therefore select the same test queries.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct ValidationWindow {
    /// Window identifier: unix_timestamp_seconds / 3600.
    pub window_id: u64,
    /// Start of the window (Unix seconds).
    pub start_secs: u64,
    /// End of the window (Unix seconds).
    pub end_secs: u64,
}

impl ValidationWindow {
    /// Compute the current validation window from the system clock.
    pub fn current() -> Self {
        let now_secs = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_secs();
        Self::from_timestamp(now_secs)
    }

    /// Compute the validation window for a given Unix timestamp (seconds).
    pub fn from_timestamp(unix_secs: u64) -> Self {
        let window_id = unix_secs / WINDOW_DURATION_SECS;
        let start_secs = window_id * WINDOW_DURATION_SECS;
        let end_secs = start_secs + WINDOW_DURATION_SECS;

        Self {
            window_id,
            start_secs,
            end_secs,
        }
    }

    /// Compute the validation window for a given window ID directly.
    pub fn from_window_id(window_id: u64) -> Self {
        let start_secs = window_id * WINDOW_DURATION_SECS;
        let end_secs = start_secs + WINDOW_DURATION_SECS;

        Self {
            window_id,
            start_secs,
            end_secs,
        }
    }

    /// Create a deterministic RNG seeded by BLAKE3(window_id).
    ///
    /// Per CP-015 section 3: all validators in the same window use the
    /// same RNG to select test queries, ensuring independent agreement.
    pub fn rng(&self) -> ChaCha20Rng {
        let seed_hash = blake3::hash(&self.window_id.to_le_bytes());
        let seed_bytes: [u8; 32] = *seed_hash.as_bytes();
        ChaCha20Rng::from_seed(seed_bytes)
    }

    /// Check if a given Unix timestamp (seconds) falls within this window.
    pub fn contains(&self, unix_secs: u64) -> bool {
        unix_secs >= self.start_secs && unix_secs < self.end_secs
    }

    /// Seconds remaining in this window from the given timestamp.
    pub fn remaining_secs(&self, unix_secs: u64) -> u64 {
        if unix_secs >= self.end_secs {
            0
        } else {
            self.end_secs - unix_secs
        }
    }

    /// Select test queries from the corpus for this window.
    ///
    /// Uses deterministic RNG so all validators independently arrive
    /// at the same test set without communication.
    pub fn select_test_queries(
        &self,
        corpus: &TestCorpus,
        count: usize,
    ) -> Vec<TestQuery> {
        select_test_queries(self.window_id, &corpus.queries, count)
    }
}

/// Select test queries from a corpus using a deterministic RNG seeded
/// by the window ID.
///
/// Per CP-015 section 4: queries are shuffled deterministically using
/// a ChaCha20Rng seeded by BLAKE3(window_id), then the first `count`
/// queries are taken.
pub fn select_test_queries(
    window_id: u64,
    query_corpus: &[TestQuery],
    count: usize,
) -> Vec<TestQuery> {
    if query_corpus.is_empty() || count == 0 {
        return Vec::new();
    }

    let seed_hash = blake3::hash(&window_id.to_le_bytes());
    let seed_bytes: [u8; 32] = *seed_hash.as_bytes();
    let mut rng = ChaCha20Rng::from_seed(seed_bytes);

    let mut shuffled: Vec<TestQuery> = query_corpus.to_vec();
    shuffled.shuffle(&mut rng);

    shuffled.truncate(count);
    shuffled
}

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

    fn make_corpus(n: usize) -> Vec<TestQuery> {
        (0..n)
            .map(|i| TestQuery {
                query_text: format!("test query {}", i),
                query_embedding: vec![i as i16; 10],
                relevant_chunk_ids: vec![Uuid::from_bytes([i as u8; 16])],
                relevance_grades: vec![1],
            })
            .collect()
    }

    #[test]
    fn test_window_from_timestamp() {
        // Timestamp 7200 seconds = 2 hours since epoch
        let w = ValidationWindow::from_timestamp(7200);
        assert_eq!(w.window_id, 2);
        assert_eq!(w.start_secs, 7200);
        assert_eq!(w.end_secs, 10800);
    }

    #[test]
    fn test_window_from_timestamp_mid_hour() {
        // 1.5 hours = 5400 seconds, should be in window 1
        let w = ValidationWindow::from_timestamp(5400);
        assert_eq!(w.window_id, 1);
        assert_eq!(w.start_secs, 3600);
        assert_eq!(w.end_secs, 7200);
    }

    #[test]
    fn test_window_from_window_id() {
        let w = ValidationWindow::from_window_id(100);
        assert_eq!(w.window_id, 100);
        assert_eq!(w.start_secs, 360000);
        assert_eq!(w.end_secs, 363600);
    }

    #[test]
    fn test_window_contains() {
        let w = ValidationWindow::from_window_id(10);
        assert!(w.contains(36000)); // Start of window
        assert!(w.contains(37800)); // Middle
        assert!(w.contains(39599)); // Just before end
        assert!(!w.contains(39600)); // End (exclusive)
        assert!(!w.contains(35999)); // Before start
    }

    #[test]
    fn test_window_remaining_secs() {
        let w = ValidationWindow::from_window_id(10);
        assert_eq!(w.remaining_secs(36000), 3600); // Start: full hour
        assert_eq!(w.remaining_secs(37800), 1800); // Middle: 30 min
        assert_eq!(w.remaining_secs(39600), 0); // End: none
        assert_eq!(w.remaining_secs(40000), 0); // Past end
    }

    #[test]
    fn test_rng_deterministic() {
        let w = ValidationWindow::from_window_id(42);
        let mut rng1 = w.rng();
        let mut rng2 = w.rng();

        // Both RNGs should produce the same sequence
        let mut corpus = make_corpus(20);
        let mut corpus2 = corpus.clone();

        corpus.shuffle(&mut rng1);
        corpus2.shuffle(&mut rng2);

        for (a, b) in corpus.iter().zip(corpus2.iter()) {
            assert_eq!(a.query_text, b.query_text);
        }
    }

    #[test]
    fn test_rng_different_windows_produce_different_sequences() {
        let w1 = ValidationWindow::from_window_id(42);
        let w2 = ValidationWindow::from_window_id(43);
        let mut rng1 = w1.rng();
        let mut rng2 = w2.rng();

        let mut corpus1 = make_corpus(20);
        let mut corpus2 = corpus1.clone();

        corpus1.shuffle(&mut rng1);
        corpus2.shuffle(&mut rng2);

        // At least one query should be in a different position
        let same_count = corpus1
            .iter()
            .zip(corpus2.iter())
            .filter(|(a, b)| a.query_text == b.query_text)
            .count();
        assert!(same_count < 20, "Different windows should produce different orderings");
    }

    #[test]
    fn test_select_test_queries_count() {
        let corpus = make_corpus(50);
        let selected = select_test_queries(100, &corpus, 10);
        assert_eq!(selected.len(), 10);
    }

    #[test]
    fn test_select_test_queries_count_exceeds_corpus() {
        let corpus = make_corpus(5);
        let selected = select_test_queries(100, &corpus, 10);
        assert_eq!(selected.len(), 5); // Can only return what exists
    }

    #[test]
    fn test_select_test_queries_empty_corpus() {
        let selected = select_test_queries(100, &[], 10);
        assert!(selected.is_empty());
    }

    #[test]
    fn test_select_test_queries_zero_count() {
        let corpus = make_corpus(10);
        let selected = select_test_queries(100, &corpus, 0);
        assert!(selected.is_empty());
    }

    #[test]
    fn test_select_test_queries_deterministic() {
        let corpus = make_corpus(30);

        let s1 = select_test_queries(42, &corpus, 10);
        let s2 = select_test_queries(42, &corpus, 10);

        assert_eq!(s1.len(), s2.len());
        for (a, b) in s1.iter().zip(s2.iter()) {
            assert_eq!(a.query_text, b.query_text);
        }
    }

    #[test]
    fn test_select_test_queries_different_windows() {
        let corpus = make_corpus(30);

        let s1 = select_test_queries(42, &corpus, 10);
        let s2 = select_test_queries(43, &corpus, 10);

        // Should select different subsets (with high probability)
        let texts1: Vec<&str> = s1.iter().map(|q| q.query_text.as_str()).collect();
        let texts2: Vec<&str> = s2.iter().map(|q| q.query_text.as_str()).collect();
        assert_ne!(texts1, texts2);
    }

    #[test]
    fn test_window_boundary_consistency() {
        // Timestamps right at the boundary should be in the correct window
        let w_before = ValidationWindow::from_timestamp(3599);
        let w_at = ValidationWindow::from_timestamp(3600);
        let w_after = ValidationWindow::from_timestamp(3601);

        assert_eq!(w_before.window_id, 0);
        assert_eq!(w_at.window_id, 1);
        assert_eq!(w_after.window_id, 1);
    }

    #[test]
    fn test_window_method_select_queries() {
        let corpus = TestCorpus {
            queries: make_corpus(30),
        };
        let window = ValidationWindow::from_window_id(42);
        let selected = window.select_test_queries(&corpus, 5);
        assert_eq!(selected.len(), 5);

        // Should match the standalone function
        let standalone = select_test_queries(42, &corpus.queries, 5);
        for (a, b) in selected.iter().zip(standalone.iter()) {
            assert_eq!(a.query_text, b.query_text);
        }
    }
}