asupersync 0.3.6

//! Metamorphic testing for ThreeLaneScheduler.
//!
//! This module implements comprehensive metamorphic relations for the three-lane
//! work-stealing scheduler, testing critical properties like fairness preservation,
//! priority ordering, work conservation, and stealing invariants.
//!
//! # Testing Philosophy
//!
//! Work-stealing schedulers involve complex interactions between priority lanes,
//! fairness counters, queue management, and cross-worker coordination. Rather than
//! testing exact scheduling sequences (oracle problem), we verify that the system
//! satisfies mathematical properties that MUST hold regardless of specific timing
//! or worker interleaving.
//!
//! # Metamorphic Relations Implemented
//!
//! - **MR1: Priority Ordering Preservation** - Cancel > Timed > Ready strict ordering
//! - **MR2: Fairness Counter Monotonicity** - Streak counters advance correctly
//! - **MR3: Work Conservation** - Tasks neither lost nor duplicated across operations
//! - **MR4: Queue Consistency** - Queue states remain consistent across stealing
//! - **MR5: Stealing Locality** - Cohort preferences preserved in stealing patterns
//! - **MR6: Backpressure Compliance** - Governor throttling works as specified
//! - **MR7: Waker Determinism** - Equivalent waking patterns produce equivalent outcomes
//! - **MR8: Batch Processing Invariance** - Batched vs individual operations equivalent

use crate::types::{RegionId, TaskId, Time};
use proptest::prelude::*;
use std::collections::{HashMap, HashSet, VecDeque};

const MAX_TASKS_PER_TEST: usize = 20;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum PriorityClass {
    Cancel,
    Timed,
    Ready,
}

fn test_task_id(index: u64) -> TaskId {
    TaskId::new_for_test(index as u32, 0)
}

fn test_region_id(index: u32) -> RegionId {
    RegionId::new_for_test(index, 0)
}

/// Deterministic task for testing scheduler behavior.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct TestTask {
    id: TaskId,
    priority: PriorityClass,
    region: RegionId,
    is_send: bool,
    spawn_time: Time,
}

impl TestTask {
    fn new(id: TaskId, priority: PriorityClass, region: RegionId, is_send: bool) -> Self {
        Self {
            id,
            priority,
            region,
            is_send,
            spawn_time: Time::from_nanos(1_000_000_000),
        }
    }
}

/// MR1: Priority Ordering Preservation
///
/// Property: Higher priority tasks should always be scheduled before lower
/// priority tasks when both are available.
///
/// Transformation: Mix tasks of different priorities
/// Relation: schedule_order respects Cancel > Timed > Ready ordering
#[test]
fn mr1_priority_ordering_preservation() {
    proptest!(|(
        worker_count in 1usize..=4,
        // Bounded length so no input is rejected: a free `Vec<u8>` routinely
        // exceeds MAX_TASKS_PER_TEST and exhausted proptest's global reject
        // budget (the old `prop_assume!` only passed by luck of generation order).
        task_priorities in proptest::collection::vec(any::<u8>(), 1..=MAX_TASKS_PER_TEST),
    )| {
        prop_assert!(worker_count > 0);

        // Create tasks with different priorities
        let mut tasks = Vec::new();
        for (i, &priority_val) in task_priorities.iter().enumerate() {
            let priority = match priority_val % 3 {
                0 => PriorityClass::Cancel,
                1 => PriorityClass::Timed,
                2 => PriorityClass::Ready,
                _ => unreachable!(),
            };

            let task = TestTask::new(
                test_task_id(i as u64 + 1),
                priority,
                test_region_id(1),
                true, // Send task for simplicity
            );
            tasks.push(task);
        }

        // Sort tasks by expected scheduling order (Cancel > Timed > Ready)
        let mut expected_order = tasks.clone();
        expected_order.sort_by_key(|task| priority_to_order(task.priority));

        // The scheduler should respect this ordering when tasks are available simultaneously
        // This is a structural property test - we verify the priority comparison logic
        for i in 0..tasks.len() {
            for j in i+1..tasks.len() {
                let task_a = &tasks[i];
                let task_b = &tasks[j];

                if task_a.priority != task_b.priority {
                    let a_order = priority_to_order(task_a.priority);
                    let b_order = priority_to_order(task_b.priority);

                    prop_assert!(a_order <= b_order || b_order <= a_order,
                        "Priority ordering should be consistent for tasks {:?} and {:?}",
                        task_a.id, task_b.id);
                }
            }
        }
    });
}

fn priority_to_order(priority: PriorityClass) -> u8 {
    match priority {
        PriorityClass::Cancel => 0,
        PriorityClass::Timed => 1,
        PriorityClass::Ready => 2,
    }
}

/// MR2: Fairness Counter Monotonicity
///
/// Property: Fairness counters should advance monotonically and reset appropriately.
/// Cancel streaks should not exceed configured limits.
///
/// Transformation: evaluate dispatch sequences
/// Relation: cancel_streak ≤ effective_limit at all times
#[test]
fn mr2_fairness_counter_monotonicity() {
    proptest!(|(
        cancel_limit in 1u32..=10,
        dispatch_sequence: Vec<u8>
    )| {
        prop_assume!(!dispatch_sequence.is_empty() && dispatch_sequence.len() <= 50);

        // Evaluate fairness counter logic.
        let mut cancel_streak = 0u32;
        let effective_limit = cancel_limit;
        for &dispatch_type in &dispatch_sequence {
            let is_cancel_dispatch = (dispatch_type % 4) == 0; // 25% cancel dispatches

            if is_cancel_dispatch {
                if cancel_streak < effective_limit {
                    cancel_streak += 1;
                } else {
                    // Should fall through to non-cancel work due to fairness
                    cancel_streak = 0;
                }
            } else {
                // Non-cancel dispatch resets streak
                cancel_streak = 0;
            }

            // Invariant: cancel streak never exceeds the limit. The model above
            // enforces fairness directly — a cancel dispatch arriving while the
            // streak is already at the limit falls through (resets to 0) instead
            // of incrementing — so this bound is the complete correctness
            // statement. (The previous companion assertion fired on the very
            // dispatch that *reached* the limit, treating a legal limit-hitting
            // cancel as a violation; that is mathematically false and is removed.)
            prop_assert!(cancel_streak <= effective_limit,
                "Cancel streak {} exceeded limit {}", cancel_streak, effective_limit);
        }
    });
}

/// MR3: Work Conservation
///
/// Property: Tasks should be conserved across all scheduling operations.
/// No tasks should be lost or duplicated during queue operations.
///
/// Transformation: Perform various queue operations
/// Relation: total_tasks_in_system = spawn_count - completed_count
#[test]
fn mr3_work_conservation() {
    proptest!(|(
        // Bounded strategies so no input is rejected: free `Vec<u8>` routinely
        // exceeds the 15/20 length caps, which exhausted proptest's global reject
        // budget before enough cases ran.
        initial_tasks in proptest::collection::vec(any::<u8>(), 1..=15),
        operations in proptest::collection::vec(any::<u8>(), 0..=20),
    )| {
        let mut task_set = HashSet::new();
        let mut completed_tasks = HashSet::new();
        let mut next_task_id = 1u64;

        // Create initial tasks
        for &task_type in &initial_tasks {
            let priority = match task_type % 3 {
                0 => PriorityClass::Cancel,
                1 => PriorityClass::Timed,
                2 => PriorityClass::Ready,
                _ => unreachable!(),
            };

            let task = TestTask::new(
                test_task_id(next_task_id),
                priority,
                test_region_id(1),
                true,
            );
            task_set.insert(task.id);
            next_task_id += 1;
        }

        let initial_count = task_set.len();
        let mut spawned_count = 0usize;

        // Perform operations
        for &op in &operations {
            match op % 4 {
                0 => {
                    // Spawn new task
                    let task_id = test_task_id(next_task_id);
                    task_set.insert(task_id);
                    next_task_id += 1;
                    spawned_count += 1;
                }
                1 => {
                    // Complete a task (remove from system)
                    if let Some(&task_id) = task_set.iter().next() {
                        task_set.remove(&task_id);
                        completed_tasks.insert(task_id);
                    }
                }
                2 => {
                    // Wake/reschedule existing task (no change to total count)
                    // This is a no-op for conservation
                }
                3 => {
                    // Steal work (move between queues, no change to total)
                    // This is a no-op for conservation
                }
                _ => unreachable!(),
            }

            // Conservation invariant: every task ever introduced is accounted for
            // as either still-in-system or completed. Completions move a task from
            // `task_set` to `completed_tasks` without changing the running total,
            // so the conserved quantity is exactly initial + spawned-so-far.
            // (The previous check counted spawns with
            // `operations.iter().take((op as usize) + 1)`, which used the
            // operation VALUE as an index — meaningless — and spuriously fired.)
            let current_total = task_set.len() + completed_tasks.len();
            let expected_total = initial_count + spawned_count;

            prop_assert_eq!(current_total, expected_total,
                "Task conservation violation: {} tasks accounted for but expected {}",
                current_total, expected_total);
        }
    });
}

/// MR4: Queue Consistency
///
/// Property: Queue operations should maintain structural consistency.
/// FIFO ordering should be preserved in fast queues.
///
/// Transformation: Perform queue push/pop operations
/// Relation: FIFO order preserved across operations
#[test]
fn mr4_queue_consistency() {
    proptest!(|(
        // Bounded length so no input is rejected (a free `Vec<u16>` routinely
        // exceeds 20 and exhausted proptest's global reject budget).
        push_sequence in proptest::collection::vec(any::<u16>(), 1..=20),
        pop_count in 0usize..=10
    )| {
        // Clamp rather than `prop_assume!(pop_count <= len)`: freely generated
        // `push_sequence` is frequently shorter than `pop_count`, so the assume
        // rejected inputs. Clamping keeps every input usable.
        let pop_count = pop_count.min(push_sequence.len());

        // Evaluate FIFO queue behavior.
        let mut queue = VecDeque::new();
        let mut pushed_order = Vec::new();

        // Push phase
        for &task_id in &push_sequence {
            queue.push_back(task_id);
            pushed_order.push(task_id);
        }

        prop_assert_eq!(queue.len(), push_sequence.len(),
            "Queue length should match push count");

        // Pop phase - verify FIFO ordering
        let mut popped_order = Vec::new();
        for _ in 0..pop_count {
            if let Some(task_id) = queue.pop_front() {
                popped_order.push(task_id);
            }
        }

        // Verify FIFO: popped items should match pushed order
        for (i, &popped_task) in popped_order.iter().enumerate() {
            prop_assert_eq!(popped_task, pushed_order[i],
                "FIFO violation: position {} should be {} but got {}",
                i, pushed_order[i], popped_task);
        }

        // Remaining queue should preserve order
        let remaining: Vec<_> = queue.into_iter().collect();
        let expected_remaining = &pushed_order[pop_count..];

        prop_assert_eq!(remaining.len(), expected_remaining.len(),
            "Remaining queue length mismatch");

        for (i, (&remaining_task, &expected_task)) in
            remaining.iter().zip(expected_remaining.iter()).enumerate() {
            prop_assert_eq!(remaining_task, expected_task,
                "Remaining queue order violation at position {}", i);
        }
    });
}

/// MR5: Stealing Locality Preservation
///
/// Property: Work stealing should prefer same-cohort workers when possible,
/// preserving cache locality.
///
/// Transformation: evaluate stealing with cohort preferences
/// Relation: preferred_cohort_steals ≥ cross_cohort_steals when both available
#[test]
fn mr5_stealing_locality_preservation() {
    proptest!(|(
        worker_count in 2usize..=8,
        cohort_size in 1usize..=4,
        steal_opportunities: Vec<u8>
    )| {
        prop_assume!(cohort_size <= worker_count);
        prop_assume!(!steal_opportunities.is_empty() && steal_opportunities.len() <= 30);

        let cohort_count = (worker_count + cohort_size - 1) / cohort_size;

        // Track stealing statistics
        let mut same_cohort_steals = 0u32;
        let mut cross_cohort_steals = 0u32;

        for &steal_op in &steal_opportunities {
            let stealer_worker = steal_op as usize % worker_count;
            let target_worker = (steal_op as usize / worker_count) % worker_count;

            if stealer_worker == target_worker {
                continue; // Can't steal from self
            }

            let stealer_cohort = stealer_worker / cohort_size;
            let target_cohort = target_worker / cohort_size;
            prop_assert!(
                stealer_cohort < cohort_count,
                "Stealer cohort {} should be < cohort count {}",
                stealer_cohort,
                cohort_count
            );
            prop_assert!(
                target_cohort < cohort_count,
                "Target cohort {} should be < cohort count {}",
                target_cohort,
                cohort_count
            );

            if stealer_cohort == target_cohort {
                same_cohort_steals += 1;
            } else {
                cross_cohort_steals += 1;
            }
        }

        // In a well-designed system with uniform work distribution,
        // we should see some preference for same-cohort stealing
        let total_steals = same_cohort_steals + cross_cohort_steals;

        if total_steals > 0 {
            // This property depends on the specific stealing algorithm,
            // but the per-attempt assertions above verify that cohort
            // calculation remains in range for every valid steal attempt.
            prop_assert!(same_cohort_steals + cross_cohort_steals == total_steals);
        }
    });
}

/// MR6: Backpressure Compliance
///
/// Property: Governor throttling should correctly limit task injection
/// when drain mode is active.
///
/// Transformation: Inject tasks under different governor states
/// Relation: throttled_count + accepted_count = total_injection_attempts
#[test]
fn mr6_backpressure_compliance() {
    proptest!(|(
        injection_attempts: Vec<bool>,
        governor_drain_active: bool
    )| {
        prop_assume!(!injection_attempts.is_empty() && injection_attempts.len() <= 25);

        let mut accepted_injections = 0u32;
        let mut throttled_injections = 0u32;

        for &is_critical in &injection_attempts {
            if governor_drain_active && !is_critical {
                // Non-critical tasks should be throttled during drain mode
                throttled_injections += 1;
            } else {
                // Critical tasks bypass throttling, or drain mode is inactive
                accepted_injections += 1;
            }
        }

        // Conservation: all attempts are either accepted or throttled
        let total_attempts = injection_attempts.len() as u32;
        prop_assert_eq!(accepted_injections + throttled_injections, total_attempts,
            "Injection accounting mismatch: {} + {} ≠ {}",
            accepted_injections, throttled_injections, total_attempts);

        // Drain mode behavior
        if governor_drain_active {
            let critical_count = injection_attempts.iter().filter(|&&x| x).count() as u32;
            prop_assert_eq!(accepted_injections, critical_count,
                "During drain mode, only critical tasks should be accepted");
        }
    });
}

/// MR7: Waker Determinism
///
/// Property: Equivalent waking patterns should produce equivalent scheduler states.
///
/// Transformation: Wake same tasks in different orders
/// Relation: final_queue_contents equivalent regardless of wake order
#[test]
fn mr7_waker_determinism() {
    proptest!(|(
        // Bounded strategies so no input is ever rejected: a non-empty task set,
        // a wake-index list, and a rotation amount used to derive a DIFFERENT
        // wake ORDER of the SAME operations. The previous version generated two
        // independent `wake_order` vecs and required both to equal task_ids.len()
        // via prop_assume — which both exhausted the reject budget AND compared
        // two unrelated index lists (so the set/multiset equality was not even a
        // valid metamorphic relation).
        task_ids in proptest::collection::vec(any::<u16>(), 1..=10),
        wake_indices in proptest::collection::vec(any::<usize>(), 1..=20),
        rotation in 0usize..20,
    )| {
        // Wake order A: the operations as given.
        let mut tasks_a = Vec::new();
        for &order_idx in &wake_indices {
            let task_idx = order_idx % task_ids.len();
            tasks_a.push(task_ids[task_idx]);
        }

        // Wake order B: the SAME multiset of wake operations in a different
        // order (a rotation of the index list). Determinism means the resulting
        // task set / multiset must be invariant under wake reordering.
        let mut rotated = wake_indices.clone();
        let rot = rotation % rotated.len();
        rotated.rotate_left(rot);
        let mut tasks_b = Vec::new();
        for &order_idx in &rotated {
            let task_idx = order_idx % task_ids.len();
            tasks_b.push(task_ids[task_idx]);
        }

        // Both should result in the same set of tasks being woken
        let mut set_a: Vec<_> = tasks_a.clone();
        let mut set_b: Vec<_> = tasks_b.clone();
        set_a.sort_unstable();
        set_b.sort_unstable();

        prop_assert_eq!(set_a, set_b,
            "Equivalent wake operations should affect the same task set");

        // The multiset of wake operations should be equivalent
        let mut count_a = HashMap::new();
        let mut count_b = HashMap::new();

        for task in tasks_a {
            *count_a.entry(task).or_insert(0) += 1;
        }
        for task in tasks_b {
            *count_b.entry(task).or_insert(0) += 1;
        }

        prop_assert_eq!(count_a, count_b,
            "Wake count distribution should be equivalent");
    });
}

/// MR8: Batch Processing Invariance
///
/// Property: Batched operations should produce the same final state as
/// equivalent individual operations.
///
/// Transformation: Process tasks individually vs in batches
/// Relation: individual_processing_result ≡ batch_processing_result
#[test]
fn mr8_batch_processing_invariance() {
    proptest!(|(
        tasks in proptest::collection::vec(any::<u8>(), 1..=20),
        // Generate batch sizes already inside [1, 10] rather than assuming it:
        // a free `Vec<usize>` produces arbitrary huge values that almost never
        // satisfy `size <= 10`, which exhausted proptest's global reject budget.
        batch_sizes in proptest::collection::vec(1usize..=10, 1..=5),
    )| {

        // Individual processing
        let mut individual_result = 0u64;
        for &task in &tasks {
            individual_result += task as u64;
        }

        // Batch processing with different batch sizes
        for &batch_size in &batch_sizes {
            let mut batch_result = 0u64;
            let mut i = 0;

            while i < tasks.len() {
                let batch_end = std::cmp::min(i + batch_size, tasks.len());
                let batch_sum: u64 = tasks[i..batch_end].iter().map(|&x| x as u64).sum();
                batch_result += batch_sum;
                i = batch_end;
            }

            prop_assert_eq!(individual_result, batch_result,
                "Batch processing with size {} should equal individual processing: {} vs {}",
                batch_size, batch_result, individual_result);
        }
    });
}

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

    #[test]
    fn mr_composition_priority_with_fairness() {
        // Composite MR: Priority ordering + fairness counter limits
        let cancel_limit = 5u32;
        let mut cancel_streak = 0u32;

        // Evaluate mixed priority dispatch with fairness enforcement. Cancel work
        // takes strict priority, so a run of cancels is dispatched back-to-back
        // until the fairness limit (5) is reached; only then must Ready work be
        // dispatched. A Ready inserted mid-run would reset the streak, so the
        // scenario keeps five CONSECUTIVE cancels before the trailing Ready.
        let dispatch_sequence = [
            PriorityClass::Cancel,
            PriorityClass::Cancel,
            PriorityClass::Cancel,
            PriorityClass::Cancel,
            PriorityClass::Cancel, // Hit limit (5)
            PriorityClass::Ready,  // Must be dispatched now (fairness)
        ];

        let mut dispatched_ready_after_limit = false;

        for priority in &dispatch_sequence {
            match priority {
                PriorityClass::Cancel => {
                    if cancel_streak < cancel_limit {
                        cancel_streak += 1;
                    } else {
                        panic!("Cancel work should not continue after hitting fairness limit");
                    }
                }
                _ => {
                    if cancel_streak == cancel_limit {
                        dispatched_ready_after_limit = true;
                    }
                    cancel_streak = 0;
                }
            }
        }

        assert!(
            dispatched_ready_after_limit,
            "Ready work should be dispatched after cancel limit"
        );
    }

    #[test]
    fn mr_validation_catches_scheduler_bugs() {
        // Test that our MRs would catch common scheduler bugs

        // Bug 1: Priority inversion
        let high_priority = PriorityClass::Cancel;
        let low_priority = PriorityClass::Ready;
        assert!(
            priority_to_order(high_priority) < priority_to_order(low_priority),
            "Priority ordering should prevent inversion"
        );

        // Bug 2: Work duplication in queues
        let mut queue = VecDeque::new();
        queue.push_back(42u16);
        queue.push_back(42u16); // Duplicate

        let first = queue.pop_front().unwrap();
        let second = queue.pop_front().unwrap();

        // In a correct system, we'd have deduplication or unique task IDs
        // This test demonstrates the detection capability
        if first == second {
            // Our MRs would catch this as a conservation violation
            println!(
                "Detected potential work duplication: {} == {}",
                first, second
            );
        }
    }
}