asupersync-conformance 0.3.4

//! Structured Concurrency Conformance Test Suite
//!
//! Tests covering scope ownership laws, lifecycle phases, and quiescence guarantees
//! that form the foundation of structured concurrency in asupersync.
//!
//! # Core Invariants Tested
//!
//! - **Cx Scope Ownership**: Context lifetime is tied to request scope
//! - **Request→Drain→Finalize**: Proper async operation lifecycle
//! - **Close⇒Quiescence**: Region close implies quiescence with no leaked obligations
//!
//! # Test IDs
//!
//! - SC-001: Context scope ownership - Cx outlives request scope
//! - SC-002: Context scope ownership - Cx cannot escape request scope
//! - SC-003: Request-drain-finalize lifecycle phases
//! - SC-004: Region close implies quiescence
//! - SC-005: No obligation leaks after region close
//! - SC-006: Nested scope ownership laws
//! - SC-007: Cancel propagation follows scope hierarchy
//! - SC-008: Resource cleanup on scope exit

use crate::{
    ConformanceTest, MpscReceiver, MpscSender, OneshotReceiver, OneshotSender, RuntimeInterface,
    TestCategory, TestMeta, TestResult, checkpoint,
};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::time::{Duration, Instant};

/// Get all structured concurrency conformance tests.
pub fn all_tests<RT: RuntimeInterface>() -> Vec<ConformanceTest<RT>> {
    vec![
        sc_001_context_scope_ownership::<RT>(),
        sc_002_context_cannot_escape_scope::<RT>(),
        sc_003_request_drain_finalize_lifecycle::<RT>(),
        sc_004_region_close_implies_quiescence::<RT>(),
        sc_005_no_obligation_leaks::<RT>(),
        sc_006_nested_scope_ownership::<RT>(),
        sc_007_cancel_propagation_hierarchy::<RT>(),
        sc_008_resource_cleanup_on_scope_exit::<RT>(),
    ]
}

/// SC-001: Context scope ownership - Cx outlives request scope
///
/// Verifies that contexts properly manage their lifetime and outlive
/// the request scope they're associated with.
pub fn sc_001_context_scope_ownership<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-001".to_string(),
            name: "Context scope ownership - lifetime management".to_string(),
            description: "Context must properly manage lifetime relative to request scope"
                .to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "scope".to_string(),
                "context".to_string(),
            ],
            expected: "Context outlives request scope without leaks".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting context scope ownership test",
                    serde_json::json!({}),
                );

                let completion_flag = Arc::new(AtomicBool::new(false));
                let completion_flag_clone = completion_flag.clone();

                // Spawn a task that creates a scoped context
                let task = rt.spawn(async move {
                    checkpoint("Creating scoped context", serde_json::json!({}));

                    // Simulate context creation within request scope
                    let context_active = Arc::new(AtomicBool::new(true));
                    let context_active_clone = context_active.clone();

                    // Start background work tied to context
                    let _background_task = rt.spawn(async move {
                        // Simulate work that depends on context being alive
                        while context_active_clone.load(Ordering::Acquire) {
                            rt.sleep(Duration::from_millis(1)).await;
                        }
                        checkpoint("Background work completed", serde_json::json!({}));
                    });

                    // Simulate request processing
                    rt.sleep(Duration::from_millis(10)).await;

                    // Context should remain valid throughout request
                    if !context_active.load(Ordering::Acquire) {
                        return Err("Context should remain active during request");
                    }

                    // End of request scope - context cleanup
                    context_active.store(false, Ordering::Release);
                    completion_flag_clone.store(true, Ordering::Release);

                    checkpoint("Request scope ended", serde_json::json!({}));
                    Ok(())
                });

                // Wait for task completion with timeout
                let result = rt.timeout(Duration::from_secs(1), task).await;

                match result {
                    Ok(Ok(())) => {
                        if completion_flag.load(Ordering::Acquire) {
                            checkpoint(
                                "Context scope ownership test passed",
                                serde_json::json!({}),
                            );
                            TestResult::passed()
                        } else {
                            TestResult::failed("Task completed but flag not set")
                        }
                    }
                    Ok(Err(e)) => {
                        checkpoint("Task failed", serde_json::json!({"error": e}));
                        TestResult::failed("Task execution failed")
                    }
                    Err(_) => {
                        checkpoint("Test timed out", serde_json::json!({}));
                        TestResult::failed("Test timed out - possible deadlock")
                    }
                }
            })
        },
    )
}

/// SC-002: Context cannot escape scope
///
/// Verifies that contexts cannot outlive their proper scope boundaries.
pub fn sc_002_context_cannot_escape_scope<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-002".to_string(),
            name: "Context cannot escape scope".to_string(),
            description: "Context must not be accessible outside its defining scope".to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "scope".to_string(),
                "context".to_string(),
            ],
            expected: "Context cannot outlive scope boundaries".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting context escape prevention test",
                    serde_json::json!({}),
                );

                let (tx, rx) = rt.oneshot();

                // Spawn task that attempts to leak context reference
                let task = rt.spawn(async move {
                    let context_id = {
                        // Inner scope where context is created
                        let context_id = Arc::new(AtomicU64::new(42));
                        context_id.load(Ordering::Acquire)
                        // Context should be cleaned up here
                    };

                    // Attempt to use context outside its scope should not cause issues
                    // In a real implementation, this would be a compile-time error
                    // Here we simulate the invariant being maintained at runtime

                    tx.send(context_id).expect("Send should succeed");
                    Ok(())
                });

                let result = rt.timeout(Duration::from_secs(1), task).await;

                match result {
                    Ok(Ok(())) => {
                        // Verify the value was transmitted (showing scope was properly managed)
                        let received = rx.await.expect("Should receive value");
                        if received == 42 {
                            checkpoint(
                                "Context escape prevention test passed",
                                serde_json::json!({}),
                            );
                            TestResult::passed()
                        } else {
                            TestResult::failed("Context value not preserved within scope")
                        }
                    }
                    _ => TestResult::failed("Context scope test failed"),
                }
            })
        },
    )
}

/// SC-003: Request-drain-finalize lifecycle phases
///
/// Tests the proper sequencing of request→drain→finalize phases in async operations.
pub fn sc_003_request_drain_finalize_lifecycle<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-003".to_string(),
            name: "Request-drain-finalize lifecycle".to_string(),
            description: "Async operations must follow proper request→drain→finalize phases"
                .to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "lifecycle".to_string(),
                "phases".to_string(),
            ],
            expected: "Operations follow request→drain→finalize sequence".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting request-drain-finalize lifecycle test",
                    serde_json::json!({}),
                );

                let phase_counter = Arc::new(AtomicUsize::new(0));
                let phase_counter_clone = phase_counter.clone();

                let task = rt.spawn(async move {
                    // Phase 1: Request
                    checkpoint("Phase 1: Request", serde_json::json!({}));
                    phase_counter_clone.store(1, Ordering::Release);

                    // Simulate some async work
                    rt.sleep(Duration::from_millis(5)).await;

                    // Phase 2: Drain
                    checkpoint("Phase 2: Drain", serde_json::json!({}));
                    phase_counter_clone.store(2, Ordering::Release);

                    // Simulate draining buffers/queues
                    rt.sleep(Duration::from_millis(5)).await;

                    // Phase 3: Finalize
                    checkpoint("Phase 3: Finalize", serde_json::json!({}));
                    phase_counter_clone.store(3, Ordering::Release);

                    Ok(())
                });

                let result = rt.timeout(Duration::from_secs(1), task).await;

                match result {
                    Ok(Ok(())) => {
                        let final_phase = phase_counter.load(Ordering::Acquire);
                        if final_phase == 3 {
                            checkpoint(
                                "Request-drain-finalize lifecycle test passed",
                                serde_json::json!({"final_phase": final_phase}),
                            );
                            TestResult::passed()
                        } else {
                            TestResult::failed(format!(
                                "Lifecycle incomplete, stopped at phase {}",
                                final_phase
                            ))
                        }
                    }
                    _ => TestResult::failed("Lifecycle test failed"),
                }
            })
        },
    )
}

/// SC-004: Region close implies quiescence
///
/// Verifies that when a region is closed, it reaches a quiescent state.
pub fn sc_004_region_close_implies_quiescence<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-004".to_string(),
            name: "Region close implies quiescence".to_string(),
            description: "Region closure must result in quiescent state".to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "region".to_string(),
                "quiescence".to_string(),
            ],
            expected: "Region close results in quiescent state".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting region close quiescence test",
                    serde_json::json!({}),
                );

                let active_tasks = Arc::new(AtomicUsize::new(0));
                let region_closed = Arc::new(AtomicBool::new(false));

                let tasks = {
                    let mut tasks = Vec::new();
                    let active_tasks_clone = active_tasks.clone();
                    let region_closed_clone = region_closed.clone();

                    // Spawn multiple tasks in a "region"
                    for i in 0..3 {
                        let active_tasks = active_tasks_clone.clone();
                        let region_closed = region_closed_clone.clone();

                        let task = rt.spawn(async move {
                            active_tasks.fetch_add(1, Ordering::SeqCst);
                            checkpoint("Task started", serde_json::json!({"task_id": i}));

                            // Simulate work until region close signal
                            while !region_closed.load(Ordering::Acquire) {
                                rt.sleep(Duration::from_millis(1)).await;
                            }

                            checkpoint("Task shutting down", serde_json::json!({"task_id": i}));
                            active_tasks.fetch_sub(1, Ordering::SeqCst);
                            Ok(())
                        });

                        tasks.push(task);
                    }
                    tasks
                };

                // Wait for all tasks to start
                while active_tasks.load(Ordering::Acquire) < 3 {
                    rt.sleep(Duration::from_millis(1)).await;
                }

                // Signal region close
                checkpoint("Closing region", serde_json::json!({}));
                region_closed.store(true, Ordering::Release);

                // Wait for quiescence (all tasks should finish)
                let quiescence_start = Instant::now();
                let mut all_completed = true;

                for task in tasks {
                    let result = rt.timeout(Duration::from_millis(500), task).await;
                    if result.is_err() {
                        all_completed = false;
                        break;
                    }
                }

                let final_active = active_tasks.load(Ordering::Acquire);

                if all_completed && final_active == 0 {
                    let quiescence_duration = quiescence_start.elapsed();
                    checkpoint(
                        "Quiescence achieved",
                        serde_json::json!({"duration_ms": quiescence_duration.as_millis()}),
                    );
                    TestResult::passed()
                } else {
                    TestResult::failed(format!(
                        "Quiescence not achieved: {} tasks still active, completion: {}",
                        final_active, all_completed
                    ))
                }
            })
        },
    )
}

/// SC-005: No obligation leaks after region close
///
/// Ensures that region closure doesn't leave behind leaked obligations.
pub fn sc_005_no_obligation_leaks<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-005".to_string(),
            name: "No obligation leaks after region close".to_string(),
            description: "Region close must not leak obligations".to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "obligations".to_string(),
                "leak".to_string(),
            ],
            expected: "No obligation leaks after region close".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting obligation leak prevention test",
                    serde_json::json!({}),
                );

                let obligation_count = Arc::new(AtomicUsize::new(0));
                let obligation_count_clone = obligation_count.clone();

                // Create a region with obligations
                let region_task = rt.spawn(async move {
                    // Create some obligations (simulated as reference counts)
                    for i in 0..5 {
                        obligation_count_clone.fetch_add(1, Ordering::SeqCst);
                        checkpoint(
                            "Created obligation",
                            serde_json::json!({"obligation_id": i}),
                        );

                        // Simulate async work that creates obligations
                        rt.sleep(Duration::from_millis(2)).await;
                    }

                    // Simulate region processing
                    rt.sleep(Duration::from_millis(10)).await;

                    // Clean up obligations before region close
                    while obligation_count_clone.load(Ordering::Acquire) > 0 {
                        let current = obligation_count_clone.fetch_sub(1, Ordering::SeqCst);
                        if current > 0 {
                            checkpoint(
                                "Cleaned obligation",
                                serde_json::json!({"obligation_id": current - 1}),
                            );
                        }
                        rt.sleep(Duration::from_millis(1)).await;
                    }

                    checkpoint("All obligations cleaned up", serde_json::json!({}));
                    Ok(())
                });

                let result = rt.timeout(Duration::from_secs(1), region_task).await;

                match result {
                    Ok(Ok(())) => {
                        let final_obligations = obligation_count.load(Ordering::Acquire);
                        if final_obligations == 0 {
                            checkpoint("No obligation leaks detected", serde_json::json!({}));
                            TestResult::passed()
                        } else {
                            TestResult::failed(format!("Leaked {} obligations", final_obligations))
                        }
                    }
                    _ => TestResult::failed("Obligation leak test failed"),
                }
            })
        },
    )
}

/// SC-006: Nested scope ownership laws
///
/// Tests that nested scopes properly maintain ownership hierarchy.
pub fn sc_006_nested_scope_ownership<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-006".to_string(),
            name: "Nested scope ownership laws".to_string(),
            description: "Nested scopes must maintain proper ownership hierarchy".to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "nested".to_string(),
                "ownership".to_string(),
            ],
            expected: "Nested scopes maintain proper hierarchy".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting nested scope ownership test",
                    serde_json::json!({}),
                );

                let outer_scope_active = Arc::new(AtomicBool::new(true));
                let inner_scope_active = Arc::new(AtomicBool::new(false));
                let test_completed = Arc::new(AtomicBool::new(false));

                let outer_scope_active_clone = outer_scope_active.clone();
                let inner_scope_active_clone = inner_scope_active.clone();
                let test_completed_clone = test_completed.clone();

                let task = rt.spawn(async move {
                    checkpoint("Entering outer scope", serde_json::json!({}));
                    // Outer scope
                    {
                        if !outer_scope_active_clone.load(Ordering::Acquire) {
                            return Err("Outer scope should be active");
                        }

                        // Inner scope
                        {
                            checkpoint("Entering inner scope", serde_json::json!({}));
                            inner_scope_active_clone.store(true, Ordering::Release);

                            // Both scopes should be active
                            if !outer_scope_active_clone.load(Ordering::Acquire)
                                || !inner_scope_active_clone.load(Ordering::Acquire)
                            {
                                return Err("Both scopes should be active");
                            }

                            rt.sleep(Duration::from_millis(5)).await;

                            checkpoint("Exiting inner scope", serde_json::json!({}));
                            inner_scope_active_clone.store(false, Ordering::Release);
                        }

                        // Inner scope should be inactive, outer still active
                        if !outer_scope_active_clone.load(Ordering::Acquire)
                            || inner_scope_active_clone.load(Ordering::Acquire)
                        {
                            return Err("Incorrect scope state after inner scope exit");
                        }

                        rt.sleep(Duration::from_millis(5)).await;

                        checkpoint("Exiting outer scope", serde_json::json!({}));
                        outer_scope_active_clone.store(false, Ordering::Release);
                    }

                    // Both scopes should be inactive
                    if outer_scope_active_clone.load(Ordering::Acquire)
                        || inner_scope_active_clone.load(Ordering::Acquire)
                    {
                        return Err("Scopes should be inactive after exit");
                    }

                    test_completed_clone.store(true, Ordering::Release);
                    checkpoint("Nested scope test completed", serde_json::json!({}));
                    Ok(())
                });

                let result = rt.timeout(Duration::from_secs(1), task).await;

                match result {
                    Ok(Ok(())) => {
                        if test_completed.load(Ordering::Acquire) {
                            TestResult::passed()
                        } else {
                            TestResult::failed("Test did not complete properly")
                        }
                    }
                    Ok(Err(e)) => {
                        TestResult::failed(format!("Nested scope ownership test failed: {}", e))
                    }
                    Err(_) => TestResult::failed("Test timed out"),
                }
            })
        },
    )
}

/// SC-007: Cancel propagation follows scope hierarchy
///
/// Verifies that cancellation properly propagates through scope hierarchy.
pub fn sc_007_cancel_propagation_hierarchy<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-007".to_string(),
            name: "Cancel propagation hierarchy".to_string(),
            description: "Cancellation must propagate correctly through scope hierarchy"
                .to_string(),
            category: TestCategory::Cancel,
            tags: vec![
                "structured".to_string(),
                "cancel".to_string(),
                "hierarchy".to_string(),
            ],
            expected: "Cancellation propagates through scope hierarchy".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint(
                    "Starting cancel propagation hierarchy test",
                    serde_json::json!({}),
                );

                let cancel_signal = Arc::new(AtomicBool::new(false));
                let child_cancelled = Arc::new(AtomicBool::new(false));

                let cancel_signal_clone = cancel_signal.clone();
                let child_cancelled_clone = child_cancelled.clone();

                // Parent scope with child task
                let parent_task = rt.spawn(async move {
                    checkpoint("Parent task started", serde_json::json!({}));

                    // Child task within parent scope
                    let child_task = rt.spawn(async move {
                        checkpoint("Child task started", serde_json::json!({}));

                        // Wait for cancel signal or work completion
                        while !cancel_signal_clone.load(Ordering::Acquire) {
                            rt.sleep(Duration::from_millis(1)).await;
                        }

                        checkpoint("Child task received cancellation", serde_json::json!({}));
                        child_cancelled_clone.store(true, Ordering::Release);
                        Ok(())
                    });

                    // Simulate some work then cancel
                    rt.sleep(Duration::from_millis(10)).await;

                    checkpoint("Parent signaling cancellation", serde_json::json!({}));
                    cancel_signal_clone.store(true, Ordering::Release);

                    // Wait for child to acknowledge cancellation
                    let child_result = rt.timeout(Duration::from_millis(100), child_task).await;

                    match child_result {
                        Ok(Ok(())) => {
                            checkpoint("Child task completed gracefully", serde_json::json!({}));
                            Ok(())
                        }
                        _ => {
                            checkpoint(
                                "Child task did not complete properly",
                                serde_json::json!({}),
                            );
                            Err("Child cancellation failed")
                        }
                    }
                });

                let result = rt.timeout(Duration::from_secs(1), parent_task).await;

                match result {
                    Ok(Ok(())) => {
                        if child_cancelled.load(Ordering::Acquire) {
                            checkpoint(
                                "Cancel propagation hierarchy test passed",
                                serde_json::json!({}),
                            );
                            TestResult::passed()
                        } else {
                            TestResult::failed("Child task was not cancelled")
                        }
                    }
                    Ok(Err(e)) => {
                        TestResult::failed(format!("Cancel propagation test failed: {}", e))
                    }
                    Err(_) => TestResult::failed("Test timed out"),
                }
            })
        },
    )
}

/// SC-008: Resource cleanup on scope exit
///
/// Ensures that resources are properly cleaned up when scopes exit.
pub fn sc_008_resource_cleanup_on_scope_exit<RT: RuntimeInterface>() -> ConformanceTest<RT> {
    ConformanceTest::new(
        TestMeta {
            id: "sc-008".to_string(),
            name: "Resource cleanup on scope exit".to_string(),
            description: "Resources must be cleaned up when scopes exit".to_string(),
            category: TestCategory::Spawn,
            tags: vec![
                "structured".to_string(),
                "cleanup".to_string(),
                "resources".to_string(),
            ],
            expected: "Resources cleaned up on scope exit".to_string(),
        },
        |rt| {
            rt.block_on(async {
                checkpoint("Starting resource cleanup test", serde_json::json!({}));

                let resource_count = Arc::new(AtomicUsize::new(0));
                let cleanup_called = Arc::new(AtomicBool::new(false));

                let resource_count_clone = resource_count.clone();
                let cleanup_called_clone = cleanup_called.clone();

                let task = rt.spawn(async move {
                    // Scope with resources
                    {
                        checkpoint("Acquiring resources", serde_json::json!({}));

                        // Simulate resource acquisition
                        for i in 0..3 {
                            resource_count_clone.fetch_add(1, Ordering::SeqCst);
                            checkpoint("Acquired resource", serde_json::json!({"resource_id": i}));
                        }

                        // Simulate work with resources
                        rt.sleep(Duration::from_millis(10)).await;

                        checkpoint(
                            "Scope ending - cleaning up resources",
                            serde_json::json!({}),
                        );

                        // Simulate resource cleanup (would be in Drop impl in real code)
                        while resource_count_clone.load(Ordering::Acquire) > 0 {
                            let current = resource_count_clone.fetch_sub(1, Ordering::SeqCst);
                            if current > 0 {
                                checkpoint(
                                    "Released resource",
                                    serde_json::json!({"resource_id": current - 1}),
                                );
                            }
                        }

                        cleanup_called_clone.store(true, Ordering::Release);
                    }
                    // Resources should be cleaned up here

                    checkpoint("Scope exited", serde_json::json!({}));
                    Ok(())
                });

                let result = rt.timeout(Duration::from_secs(1), task).await;

                match result {
                    Ok(Ok(())) => {
                        let remaining_resources = resource_count.load(Ordering::Acquire);
                        let cleanup_performed = cleanup_called.load(Ordering::Acquire);

                        if remaining_resources == 0 && cleanup_performed {
                            checkpoint("Resource cleanup test passed", serde_json::json!({}));
                            TestResult::passed()
                        } else {
                            TestResult::failed(format!(
                                "Resource cleanup failed: {} remaining, cleanup called: {}",
                                remaining_resources, cleanup_performed
                            ))
                        }
                    }
                    _ => TestResult::failed("Resource cleanup test failed"),
                }
            })
        },
    )
}