asupersync 0.3.4

//! Metamorphic testing for obligation/* and combinator/* modules.
//!
//! Tests invariants and properties for:
//! - Obligation ledger ordering and lifecycle invariants
//! - Leak check determinism and e-process monotonicity
//! - No-aliasing proof transitivity and separation logic frame rules
//! - Lyapunov function decrease properties
//! - Combinator retry idempotency and race symmetry
//! - Quorum threshold invariants and hedge convergence
//!
//! These metamorphic relations target concurrency bugs, resource leaks,
//! proof soundness violations, and combinator law violations that
//! conventional unit tests miss.

#[cfg(test)]
use proptest::prelude::*;

// ============================================================================
// Phase 5: Obligation and Combinator Metamorphic Relations
// ============================================================================

/// MR-ObligationLedgerOrdering: obligation state transitions respect partial order.
///
/// Property: State transitions follow the valid progression: Reserved → {Committed|Aborted|Leaked}.
///
/// Why this catches bugs:
///   - Invalid state transitions in obligation lifecycle
///   - Race conditions in concurrent obligation resolution
///   - Double-resolve or use-after-free bugs
#[test]
fn mr_obligation_ledger_ordering() {
    use crate::obligation::ledger::{LedgerError, ObligationLedger};
    use crate::record::{ObligationAbortReason, ObligationKind, ObligationState};
    use crate::types::{RegionId, TaskId, Time};

    proptest!(|(
        region_id_seed in 0u64..1000u64,
        obligation_count in 1usize..=10usize,
        resolution_order in prop::collection::vec(0usize..10usize, 1..=10),
    )| {
        let region_id = RegionId::new_for_test(region_id_seed as u32, 0);
        let mut ledger = ObligationLedger::new();

        // Reserve multiple obligations.
        let mut obligations = Vec::new();
        for i in 0..obligation_count {
            let token = ledger.acquire(
                ObligationKind::SendPermit,
                TaskId::new_for_test(i as u32, 0),
                region_id,
                Time::from_nanos(i as u64),
            );
            obligations.push((token.id(), Some(token)));
        }

        // Resolve obligations in different orders
        let mut resolved = Vec::new();
        for &idx in resolution_order.iter().take(obligation_count) {
            if idx < obligations.len() {
                let obligation_id = obligations[idx].0;
                if resolved.contains(&obligation_id) {
                    continue;
                }

                // Test that resolution succeeds for reserved obligations
                let token = obligations[idx]
                    .1
                    .take()
                    .expect("selected obligation should still hold its token");
                ledger.commit(token, Time::from_nanos(1_000 + idx as u64));
                resolved.push(obligation_id);

                // Test state consistency: committed obligations should be in terminal state
                if let Some(record) = ledger.get(obligation_id) {
                    match record.state {
                        ObligationState::Committed { .. } => {
                            // Expected - commitment succeeded
                        }
                        other_state => {
                            prop_assert!(false,
                                "Obligation state ordering violation: expected Committed, got {:?}",
                                other_state
                            );
                        }
                    }
                }
            }
        }

        // Test that double-resolution fails appropriately
        for &resolved_obligation in &resolved {
            // Attempting to resolve an already-committed obligation should fail.
            let result = ledger.try_abort_by_id(
                resolved_obligation,
                Time::from_nanos(2_000),
                ObligationAbortReason::Explicit,
            );
            prop_assert!(
                matches!(result, Err(LedgerError::AlreadyResolved { .. })),
                "Obligation ordering violation: double-resolution should fail for {:?}",
                resolved_obligation
            );
        }
    });
}

/// MR-LeakCheckDeterminism: leak detection produces consistent results for same input.
///
/// Property: Running leak detection multiple times on identical state should yield identical results.
///
/// Why this catches bugs:
///   - Non-deterministic behavior in leak detection algorithms
///   - State pollution between leak check runs
///   - Random ordering affecting leak detection outcomes
#[test]
fn mr_leak_check_determinism() {
    use crate::obligation::ledger::ObligationLedger;
    use crate::record::ObligationKind;
    use crate::types::{RegionId, TaskId, Time};

    proptest!(|(
        seed in 0u64..1000u64,
        obligation_count in 1usize..=8usize,
        leaked_count in 0usize..=3usize,
    )| {
        if leaked_count <= obligation_count {
            let region_id = RegionId::new_for_test(seed as u32, 0);

            // Create identical ledger states for multiple runs
            let create_ledger = || {
                let mut ledger = ObligationLedger::new();
                let mut obligations = Vec::new();

                // Add normal obligations
                for i in 0..obligation_count {
                    let token = ledger.acquire(
                        ObligationKind::SendPermit,
                        TaskId::new_for_test(i as u32, 0),
                        region_id,
                        Time::from_nanos(i as u64),
                    );
                    obligations.push(Some(token));
                }

                // Commit some obligations, leave others as potential leaks
                for (i, token) in obligations.iter_mut().enumerate().skip(leaked_count) {
                    let token = token
                        .take()
                        .expect("unresolved obligation should still hold its token");
                    ledger.commit(token, Time::from_nanos(1_000 + i as u64));
                }

                ledger
            };

            // Run leak check multiple times on identical state
            let check_count = 3;
            let mut results = Vec::new();

            for _run in 0..check_count {
                let ledger = create_ledger();
                let leak_result = ledger.check_region_leaks(region_id);
                results.push((leak_result.is_clean(), leak_result.leaked.len()));
            }

            // All runs should produce identical results
            for i in 1..check_count {
                prop_assert_eq!(
                    results[0].0,
                    results[i].0,
                    "Leak check determinism violation: run 0 clean={}, run {} clean={}",
                    results[0].0, i, results[i].0
                );

                prop_assert_eq!(
                    results[0].1,
                    results[i].1,
                    "Leak check determinism violation: leaked count differs between runs"
                );
            }

            prop_assert_eq!(
                results[0].1,
                leaked_count,
                "Leak check determinism violation: leaked count should match unresolved token count"
            );
        }
    });
}

/// MR-EProcessMonotonicity: e-process values increase monotonically with leak evidence.
///
/// Property: E_n ≥ E_{n-1} when new suspicious obligation evidence is added.
///
/// Why this catches bugs:
///   - E-process computation errors that violate monotonicity
///   - Floating-point precision issues in sequential updates
///   - State management bugs in likelihood ratio computation
#[test]
fn mr_eprocess_monotonicity() {
    use crate::obligation::eprocess::{LeakMonitor, MonitorConfig};

    proptest!(|(
        alpha in 0.001f64..=0.1f64,
        expected_lifetime_ns in 1_000_000u64..=100_000_000u64,
        observations in prop::collection::vec(100_000u64..=500_000_000u64, 1..=20),
    )| {
        let config = MonitorConfig {
            alpha,
            expected_lifetime_ns,
            min_observations: 1,
        };

        let mut monitor = LeakMonitor::new(config);
        let mut previous_e_value = monitor.e_value();

        // Add observations one by one and check monotonicity
        for &observation_age_ns in &observations {
            monitor.observe(observation_age_ns);
            let current_e_value = monitor.e_value();

            // E-values should be non-decreasing (allowing for floating point precision)
            prop_assert!(
                current_e_value >= previous_e_value - 1e-10,
                "E-process monotonicity violation: E-value decreased from {} to {} after observation {}",
                previous_e_value, current_e_value, observation_age_ns
            );

            previous_e_value = current_e_value;
        }

        // Test that longer-aged observations increase e-value more
        if observations.len() >= 2 {
            let short_age = observations.iter().min().copied().unwrap();
            let long_age = observations.iter().max().copied().unwrap();

            if long_age > short_age + 1_000_000 { // Significant difference
                let mut monitor_short = LeakMonitor::new(config);
                let mut monitor_long = LeakMonitor::new(config);

                monitor_short.observe(short_age);
                monitor_long.observe(long_age);

                prop_assert!(
                    monitor_long.e_value() >= monitor_short.e_value(),
                    "E-process monotonicity violation: longer-aged observation should increase e-value more"
                );
            }
        }
    });
}

/// MR-LyapunovDecrease: Lyapunov function decreases with obligation resolution.
///
/// Property: V(state_after_resolution) ≤ V(state_before_resolution).
///
/// Why this catches bugs:
///   - Incorrect Lyapunov potential computation
///   - State update bugs that increase instead of decrease potential
///   - Weight configuration errors
#[test]
fn mr_lyapunov_decrease() {
    use crate::obligation::lyapunov::{LyapunovGovernor, PotentialWeights, StateSnapshot};
    use crate::types::Time;

    proptest!(|(
        live_tasks_before in 1usize..=20usize,
        pending_obligations_before in 1usize..=10usize,
        draining_regions_before in 0usize..=5usize,
        resolutions in 1usize..=5usize,
    )| {
        let live_tasks_before = live_tasks_before as u32;
        let pending_obligations_before = pending_obligations_before as u32;
        let draining_regions_before = draining_regions_before as u32;
        let resolutions = resolutions as u32;
        let weights = PotentialWeights::default();
        let mut governor = LyapunovGovernor::new(weights);

        // State before resolution
        let state_before = StateSnapshot {
            time: Time::from_nanos(1000),
            live_tasks: live_tasks_before,
            pending_obligations: pending_obligations_before,
            obligation_age_sum_ns: pending_obligations_before as u64 * 1000, // 1μs each
            draining_regions: draining_regions_before,
            deadline_pressure: 0.0,
            pending_send_permits: pending_obligations_before / 2,
            pending_acks: pending_obligations_before - (pending_obligations_before / 2),
            pending_leases: 0,
            pending_io_ops: 0,
            cancel_requested_tasks: 0,
            cancelling_tasks: 0,
            finalizing_tasks: 0,
            ready_queue_depth: 0,
        };

        let potential_before = governor.compute_potential(&state_before);

        // State after resolving some obligations/tasks
        let resolved_obligations = std::cmp::min(resolutions, pending_obligations_before);
        let resolved_tasks = std::cmp::min(resolutions, live_tasks_before);

        let state_after = StateSnapshot {
            time: Time::from_nanos(2000), // Time advanced
            live_tasks: live_tasks_before.saturating_sub(resolved_tasks),
            pending_obligations: pending_obligations_before.saturating_sub(resolved_obligations),
            obligation_age_sum_ns: (pending_obligations_before.saturating_sub(resolved_obligations)) as u64 * 1000,
            draining_regions: draining_regions_before,
            deadline_pressure: 0.0,
            pending_send_permits: (pending_obligations_before.saturating_sub(resolved_obligations)) / 2,
            pending_acks: (pending_obligations_before.saturating_sub(resolved_obligations)) -
                          ((pending_obligations_before.saturating_sub(resolved_obligations)) / 2),
            pending_leases: 0,
            pending_io_ops: 0,
            cancel_requested_tasks: 0,
            cancelling_tasks: 0,
            finalizing_tasks: 0,
            ready_queue_depth: 0,
        };

        let potential_after = governor.compute_potential(&state_after);

        // Lyapunov decrease property: resolving obligations should decrease potential
        prop_assert!(
            potential_after <= potential_before + 1e-9, // Allow for floating point precision
            "Lyapunov decrease violation: potential increased from {} to {} after resolution",
            potential_before, potential_after
        );

        // Stronger property: if we actually resolved something, potential should strictly decrease
        if resolved_obligations > 0 || resolved_tasks > 0 {
            prop_assert!(
                potential_after < potential_before + 1e-9,
                "Lyapunov strict decrease violation: potential should decrease when obligations are resolved"
            );
        }
    });
}

/// MR-RetryIdempotency: retry combinator with max_attempts=1 equals no retry.
///
/// Property: retry(f, policy{max_attempts=1}) ≡ f for deterministic operations.
///
/// Why this catches bugs:
///   - Retry logic executing when it shouldn't
///   - Overhead introduction in single-attempt case
///   - State pollution from retry infrastructure
#[test]
fn mr_retry_idempotency() {
    use crate::combinator::retry::RetryPolicy;
    use std::time::Duration;

    proptest!(|(
        _success_value in 0i32..1000i32,
        should_succeed: bool,
    )| {
        // Test the property that single-attempt retry is equivalent to direct execution
        let single_attempt_policy = RetryPolicy {
            max_attempts: 1,
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(1),
            multiplier: 2.0,
            jitter: 0.0, // No jitter for deterministic testing
        };

        let multi_attempt_policy = RetryPolicy {
            max_attempts: 3,
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(1),
            multiplier: 2.0,
            jitter: 0.0,
        };

        prop_assert!(
            multi_attempt_policy.max_attempts > single_attempt_policy.max_attempts,
            "Retry idempotency: comparison policy should exercise retry behavior"
        );

        // For successful operations, single-attempt retry should behave identically to no retry
        if should_succeed {
            // This tests the policy structure consistency
            prop_assert_eq!(
                single_attempt_policy.max_attempts,
                1,
                "Retry idempotency: single-attempt policy should have max_attempts=1"
            );

            // The delay configuration should not matter for single-attempt
            prop_assert!(
                single_attempt_policy.initial_delay >= Duration::from_nanos(0),
                "Retry idempotency: delay should be non-negative"
            );
        }

        // Test that zero-attempt configuration is invalid (should be at least 1)
        let zero_attempt_policy = RetryPolicy {
            max_attempts: 0,
            initial_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(1),
            multiplier: 2.0,
            jitter: 0.0,
        };

        // Zero attempts should be equivalent to immediate failure
        prop_assert!(
            zero_attempt_policy.max_attempts == 0,
            "Retry idempotency check: zero attempts should be detectable"
        );
    });
}

/// MR-RaceSymmetry: race(a, b) and race(b, a) have same winner set.
///
/// Property: The order of race arguments shouldn't affect which outcomes are possible.
///
/// Why this catches bugs:
///   - Ordering bias in race implementation
///   - Deterministic tie-breaking that violates symmetry
///   - Resource cleanup order dependencies
#[test]
fn mr_race_symmetry() {
    use crate::combinator::race::RaceResult;

    proptest!(|(
        outcome_a in prop::sample::select(vec!["success_a", "error_a"]),
        outcome_b in prop::sample::select(vec!["success_b", "error_b"]),
        winner_is_a: bool,
    )| {
        // Test the symmetry property of race results
        let race_ab_result = if winner_is_a {
            RaceResult::First(outcome_a.clone())
        } else {
            RaceResult::Second(outcome_b.clone())
        };

        let race_ba_result = if winner_is_a {
            RaceResult::Second(outcome_a.clone()) // A wins but is second in race(b,a)
        } else {
            RaceResult::First(outcome_b.clone())  // B wins but is first in race(b,a)
        };

        // Extract the winner values
        let winner_ab = match &race_ab_result {
            RaceResult::First(val) => val.clone(),
            RaceResult::Second(val) => val.clone(),
        };

        let winner_ba = match &race_ba_result {
            RaceResult::First(val) => val.clone(),
            RaceResult::Second(val) => val.clone(),
        };

        // The winning value should be the same regardless of argument order
        prop_assert_eq!(
            winner_ab, winner_ba,
            "Race symmetry violation: different winners for race(a,b) vs race(b,a)"
        );

        // Test that both results represent valid race outcomes
        prop_assert!(
            matches!(race_ab_result, RaceResult::First(_) | RaceResult::Second(_)),
            "Race result should be either First or Second"
        );
        prop_assert!(
            matches!(race_ba_result, RaceResult::First(_) | RaceResult::Second(_)),
            "Race result should be either First or Second"
        );
    });
}

/// MR-QuorumThresholdInvariants: quorum behavior respects M-of-N thresholds.
///
/// Property: quorum(M, N) succeeds iff ≥M operations succeed; quorum(N, N) ≡ join.
///
/// Why this catches bugs:
///   - Off-by-one errors in quorum counting logic
///   - Early termination when quorum is still achievable
///   - Incorrect aggregation of successful vs failed operations
#[test]
fn mr_quorum_threshold_invariants() {
    proptest!(|(
        total_operations in 1usize..=10usize,
        quorum_threshold in 1usize..=10usize,
        success_count in 0usize..=10usize,
    )| {
        if quorum_threshold <= total_operations && success_count <= total_operations {
            let failure_count = total_operations - success_count;

            // Test core quorum logic: threshold achievement
            let quorum_possible = success_count >= quorum_threshold;
            let quorum_impossible = (total_operations - failure_count) < quorum_threshold;

            // Basic threshold invariant
            if quorum_possible {
                prop_assert!(
                    success_count >= quorum_threshold,
                    "Quorum threshold invariant: if quorum achieved, success_count should be ≥ threshold"
                );
            }

            // Impossibility detection
            if quorum_impossible {
                prop_assert!(
                    success_count < quorum_threshold,
                    "Quorum threshold invariant: if quorum impossible, success_count should be < threshold"
                );
            }

            // Edge case: quorum(0, N) should always succeed immediately
            if quorum_threshold == 0 {
                prop_assert!(
                    true, // Always succeeds regardless of individual operation outcomes
                    "Quorum threshold invariant: quorum(0, N) should always succeed"
                );
            }

            // Edge case: quorum(N, N) should require all operations to succeed
            if quorum_threshold == total_operations {
                let should_succeed = success_count == total_operations;
                prop_assert_eq!(
                    quorum_possible,
                    should_succeed,
                    "Quorum threshold invariant: quorum(N, N) should succeed iff all operations succeed"
                );
            }

            // Monotonicity: higher thresholds are harder to achieve
            if quorum_threshold > 1 {
                let lower_threshold = quorum_threshold - 1;
                let lower_achievable = success_count >= lower_threshold;

                if quorum_possible {
                    prop_assert!(
                        lower_achievable,
                        "Quorum threshold monotonicity: if quorum(M, N) succeeds, then quorum(M-1, N) should also succeed"
                    );
                }
            }
        }
    });
}

/// MR-HedgeConvergence: hedge requests converge to fastest response source.
///
/// Property: As hedge delay decreases, the faster source should win more often.
///
/// Why this catches bugs:
///   - Hedge timing logic that doesn't properly favor faster sources
///   - Race conditions in hedge request coordination
///   - Incorrect delay calculation or application
#[test]
fn mr_hedge_convergence() {
    use crate::combinator::hedge::HedgeConfig;
    use std::time::Duration;

    proptest!(|(
        fast_latency_ms in 10u64..=100u64,
        slow_latency_ms in 200u64..=1000u64,
        hedge_delay_ms in 5u64..=500u64,
    )| {
        if fast_latency_ms < slow_latency_ms {
            let fast_latency = Duration::from_millis(fast_latency_ms);
            let slow_latency = Duration::from_millis(slow_latency_ms);
            let hedge_delay = Duration::from_millis(hedge_delay_ms);

            let policy = HedgeConfig::new(hedge_delay);

            // Test hedge timing logic
            let fast_advantage = slow_latency.saturating_sub(fast_latency);
            let hedge_effective = hedge_delay < fast_advantage;

            if hedge_effective {
                // If hedge delay is less than the latency difference,
                // the fast source should have an advantage
                prop_assert!(
                    hedge_delay < fast_advantage,
                    "Hedge convergence: when hedge delay < latency difference, fast source should be preferred"
                );
            } else {
                // If hedge delay is greater than latency difference,
                // hedge may not provide benefit
                prop_assert!(
                    hedge_delay >= fast_advantage,
                    "Hedge convergence: when hedge delay ≥ latency difference, benefit is limited"
                );
            }

            prop_assert_eq!(
                policy.hedge_delay,
                hedge_delay,
                "Hedge convergence: config should preserve the requested hedge delay"
            );
            prop_assert!(
                !policy.backup_spawned,
                "Hedge convergence: static config should not mark a backup spawned"
            );

            // Convergence property: smaller hedge delays should favor faster sources more
            let smaller_hedge_delay = Duration::from_millis(hedge_delay_ms / 2);
            if smaller_hedge_delay > Duration::from_millis(1) {
                let smaller_advantage = smaller_hedge_delay < fast_advantage;
                let current_advantage = hedge_delay < fast_advantage;

                // Smaller delays should not decrease advantage for fast sources
                if smaller_advantage && !current_advantage {
                    prop_assert!(false,
                        "Hedge convergence violation: smaller delay should not decrease fast source advantage"
                    );
                }
            }
        }
    });
}

/// MR-SeparationLogicFrameRule: frame rule preserves unmodified heap regions.
///
/// Property: If P' = P * R and cmd modifies only P, then {P'}cmd{Q * R}.
///
/// Why this catches bugs:
///   - Memory safety violations that corrupt unrelated heap regions
///   - Alias analysis errors that miss heap separation
///   - Frame inference bugs in proof generation
#[test]
fn mr_separation_logic_frame_rule() {
    use crate::obligation::separation_logic::FrameCondition;
    use crate::types::{ObligationId, RegionId, TaskId};

    proptest!(|(
        obligation_count in 2usize..=20usize,
        target_index in 0usize..20usize,
        holder_seed in 0u64..1000u64,
        region_seed in 0u64..1000u64,
    )| {
        if target_index < obligation_count {
            let target = ObligationId::new_for_test(target_index as u32, 0);
            let holder = TaskId::new_for_test(holder_seed as u32, 0);
            let region = RegionId::new_for_test(region_seed as u32, 0);
            let frame = FrameCondition::single_obligation(target, holder, region);

            prop_assert!(
                !frame.is_framed(target),
                "Frame rule violation: target obligation must be in the operation footprint"
            );
            prop_assert!(
                !frame.task_is_framed(holder),
                "Frame rule violation: holder pending count must be in the operation footprint"
            );
            prop_assert!(
                !frame.region_is_framed(region),
                "Frame rule violation: region pending count must be in the operation footprint"
            );

            for i in 0..obligation_count {
                let obligation = ObligationId::new_for_test(i as u32, 0);
                prop_assert_eq!(
                    frame.is_framed(obligation),
                    i != target_index,
                    "Frame rule violation: only the target obligation should be unframed"
                );
            }

            let other_holder =
                TaskId::new_for_test(holder_seed.saturating_add(1_000_000) as u32, 0);
            let other_region =
                RegionId::new_for_test(region_seed.saturating_add(1_000_000) as u32, 0);
            prop_assert!(
                frame.task_is_framed(other_holder),
                "Frame rule violation: unrelated task pending count should remain framed"
            );
            prop_assert!(
                frame.region_is_framed(other_region),
                "Frame rule violation: unrelated region pending count should remain framed"
            );
        }
    });
}