asupersync 0.3.4

//! Actor Conformance Test Harness
//!
//! Implements Pattern 4 (Spec-Derived Test Matrix) to verify actor contracts
//! against the region-owned message-driven concurrency specification. Tests cover:
//!
//! - Actor trait implementation contracts
//! - Region ownership and structured lifecycle
//! - Sequential message handling with exclusive state access
//! - Lifecycle management (on_start/on_stop hooks)
//! - Two-phase messaging (reserve/send pattern)
//! - State transition atomicity (Created/Running/Stopping/Stopped)
//! - Handle operations and semantics
//! - ActorRef cloning and lightweight references
//! - Join semantics and completion handling
//! - Graceful stop vs immediate abort behavior
//! - Drop abort safety for cleanup guarantee

#![allow(dead_code, clippy::vec_init_then_push)]

use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll, Waker};

use crate::actor::Actor;
use crate::cx::Cx;
use crate::types::{Outcome, Time};
use futures_lite::future::block_on;
use serde_json::json;

/// Test verdict for conformance checks.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TestVerdict {
    Pass,
    Fail(String),
}

/// Test result with metadata.
#[derive(Debug, Clone)]
pub struct ConformanceTestResult {
    pub test_name: &'static str,
    pub requirement_level: RequirementLevel,
    pub category: TestCategory,
    pub verdict: TestVerdict,
}

/// RFC-style requirement levels for coverage tracking.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RequirementLevel {
    Must,   // MUST comply
    Should, // SHOULD comply
    May,    // MAY implement
}

/// Test categories for organizational purposes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TestCategory {
    ActorTraitContract,
    RegionOwnership,
    MessageHandling,
    LifecycleManagement,
    TwoPhaseMessaging,
    StateTransitions,
    HandleOperations,
    ActorRefCloning,
    JoinSemantics,
    GracefulStop,
    AbortSemantics,
    DropAbortSafety,
}

/// Deterministic actor for controlled testing.
#[derive(Debug)]
struct DeterministicActor {
    name: String,
    message_count: Arc<AtomicU64>,
    messages_received: Arc<Mutex<Vec<String>>>,
    on_start_called: Arc<AtomicBool>,
    on_stop_called: Arc<AtomicBool>,
    should_panic_in_handle: Arc<AtomicBool>,
    should_panic_in_start: Arc<AtomicBool>,
    should_panic_in_stop: Arc<AtomicBool>,
    handle_delay_ms: Arc<AtomicU64>,
}

impl DeterministicActor {
    fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            message_count: Arc::new(AtomicU64::new(0)),
            messages_received: Arc::new(Mutex::new(Vec::new())),
            on_start_called: Arc::new(AtomicBool::new(false)),
            on_stop_called: Arc::new(AtomicBool::new(false)),
            should_panic_in_handle: Arc::new(AtomicBool::new(false)),
            should_panic_in_start: Arc::new(AtomicBool::new(false)),
            should_panic_in_stop: Arc::new(AtomicBool::new(false)),
            handle_delay_ms: Arc::new(AtomicU64::new(0)),
        }
    }

    fn configure_panic(&self, handle: bool, start: bool, stop: bool) {
        self.should_panic_in_handle.store(handle, Ordering::SeqCst);
        self.should_panic_in_start.store(start, Ordering::SeqCst);
        self.should_panic_in_stop.store(stop, Ordering::SeqCst);
    }

    fn set_handle_delay(&self, delay_ms: u64) {
        self.handle_delay_ms.store(delay_ms, Ordering::SeqCst);
    }

    fn message_count(&self) -> u64 {
        self.message_count.load(Ordering::SeqCst)
    }

    fn messages_received(&self) -> Vec<String> {
        self.messages_received.lock().unwrap().clone()
    }

    fn was_on_start_called(&self) -> bool {
        self.on_start_called.load(Ordering::SeqCst)
    }

    fn was_on_stop_called(&self) -> bool {
        self.on_stop_called.load(Ordering::SeqCst)
    }
}

impl Actor for DeterministicActor {
    type Message = String;

    fn on_start(&mut self, _cx: &Cx) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            assert!(
                !self.should_panic_in_start.load(Ordering::SeqCst),
                "Intentional panic in on_start"
            );
            self.on_start_called.store(true, Ordering::SeqCst);
        })
    }

    fn handle(
        &mut self,
        _cx: &Cx,
        msg: Self::Message,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            assert!(
                !self.should_panic_in_handle.load(Ordering::SeqCst),
                "Intentional panic in handle"
            );

            let delay_ms = self.handle_delay_ms.load(Ordering::SeqCst);
            if delay_ms > 0 {
                // Exercise processing delay.
                for _ in 0..delay_ms {
                    // Busy wait for testing
                }
            }

            self.message_count.fetch_add(1, Ordering::SeqCst);
            self.messages_received.lock().unwrap().push(msg);
        })
    }

    fn on_stop(&mut self, _cx: &Cx) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            assert!(
                !self.should_panic_in_stop.load(Ordering::SeqCst),
                "Intentional panic in on_stop"
            );
            self.on_stop_called.store(true, Ordering::SeqCst);
        })
    }
}

/// Simple counter actor for basic testing.
#[derive(Debug)]
struct CounterActor {
    count: u64,
    lifecycle_events: Arc<Mutex<Vec<String>>>,
}

impl CounterActor {
    fn new() -> Self {
        Self {
            count: 0,
            lifecycle_events: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn lifecycle_events(&self) -> Vec<String> {
        self.lifecycle_events.lock().unwrap().clone()
    }
}

impl Actor for CounterActor {
    type Message = u64;

    fn on_start(&mut self, _cx: &Cx) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            self.lifecycle_events
                .lock()
                .unwrap()
                .push("on_start".into());
        })
    }

    fn handle(
        &mut self,
        _cx: &Cx,
        msg: Self::Message,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            self.count += msg;
            self.lifecycle_events
                .lock()
                .unwrap()
                .push(format!("handle({})", msg));
        })
    }

    fn on_stop(&mut self, _cx: &Cx) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(async move {
            self.lifecycle_events.lock().unwrap().push("on_stop".into());
        })
    }
}

/// Deterministic time for deterministic testing.
#[derive(Debug, Clone)]
struct DeterministicTime {
    current: Arc<Mutex<Time>>,
}

impl DeterministicTime {
    fn new() -> Self {
        Self {
            current: Arc::new(Mutex::new(Time::from_nanos(0))),
        }
    }

    fn advance_ms(&self, milliseconds: u64) {
        let mut current = self.current.lock().unwrap();
        *current = current.saturating_add_nanos(milliseconds.saturating_mul(1_000_000));
    }

    fn now(&self) -> Time {
        *self.current.lock().unwrap()
    }
}

#[derive(Debug, Clone)]
struct ActorObservation {
    observed_events: Vec<String>,
    post_stop_send_rejected: Option<bool>,
}

fn noop_waker() -> Waker {
    Waker::noop().clone()
}

fn drive_counter_actor(
    messages: &[u64],
    stop_before_run: bool,
    attempt_post_stop_send: bool,
) -> Result<ActorObservation, String> {
    let mut runtime = crate::lab::LabRuntime::new(crate::lab::LabConfig::default());
    let region = runtime
        .state
        .create_root_region(crate::types::Budget::INFINITE);
    let cx = Cx::for_testing();
    let scope = crate::cx::Scope::<crate::types::policy::FailFast>::new(
        region,
        crate::types::Budget::INFINITE,
    );

    let actor = CounterActor::new();
    let event_log = Arc::clone(&actor.lifecycle_events);
    let (handle, stored) = scope
        .spawn_actor(&mut runtime.state, &cx, actor, 32)
        .map_err(|err| format!("spawn_actor failed: {err:?}"))?;
    let task_id = handle.task_id();
    runtime.state.store_spawned_task(task_id, stored);

    for &message in messages {
        handle
            .try_send(message)
            .map_err(|err| format!("try_send({message}) failed before run: {err:?}"))?;
    }

    let mut post_stop_send_rejected = None;
    if stop_before_run {
        handle.stop();
        if attempt_post_stop_send {
            post_stop_send_rejected = Some(matches!(
                handle.try_send(999_999),
                Err(crate::channel::mpsc::SendError::Disconnected(999_999))
            ));
        }
    } else {
        drop(handle);
    }

    runtime.scheduler.lock().schedule(task_id, 0);
    runtime.run_until_quiescent();

    Ok(ActorObservation {
        observed_events: event_log.lock().unwrap().clone(),
        post_stop_send_rejected,
    })
}

fn drive_two_phase_send_actor() -> Result<Vec<String>, String> {
    let mut runtime = crate::lab::LabRuntime::new(crate::lab::LabConfig::default());
    let region = runtime
        .state
        .create_root_region(crate::types::Budget::INFINITE);
    let cx = Cx::for_testing();
    let scope = crate::cx::Scope::<crate::types::policy::FailFast>::new(
        region,
        crate::types::Budget::INFINITE,
    );

    let actor = CounterActor::new();
    let event_log = Arc::clone(&actor.lifecycle_events);
    let (handle, stored) = scope
        .spawn_actor(&mut runtime.state, &cx, actor, 32)
        .map_err(|err| format!("spawn_actor failed: {err:?}"))?;
    let task_id = handle.task_id();
    runtime.state.store_spawned_task(task_id, stored);

    let actor_ref = handle.sender();
    let permit =
        block_on(actor_ref.reserve(&cx)).map_err(|err| format!("reserve phase failed: {err:?}"))?;
    let reserved_snapshot = permit.telemetry_snapshot(0xA7C0);
    match permit.send(41) {
        Outcome::Ok(()) => {}
        other => return Err(format!("permit send phase failed: {other:?}")),
    }

    drop(actor_ref);
    drop(handle);
    runtime.scheduler.lock().schedule(task_id, 0);
    runtime.run_until_quiescent();

    let mut observed_events = vec![
        format!(
            "reserved_uncommitted_obligations={}",
            reserved_snapshot.reserved_uncommitted_obligations
        ),
        "permit_send=ok".to_string(),
    ];
    observed_events.extend(event_log.lock().unwrap().iter().cloned());
    Ok(observed_events)
}

fn drive_cancelled_reserve_actor() -> Result<Vec<String>, String> {
    let mut runtime = crate::lab::LabRuntime::new(crate::lab::LabConfig::default());
    let region = runtime
        .state
        .create_root_region(crate::types::Budget::INFINITE);
    let cx = Cx::for_testing();
    let scope = crate::cx::Scope::<crate::types::policy::FailFast>::new(
        region,
        crate::types::Budget::INFINITE,
    );

    let actor = CounterActor::new();
    let event_log = Arc::clone(&actor.lifecycle_events);
    let (handle, stored) = scope
        .spawn_actor(&mut runtime.state, &cx, actor, 1)
        .map_err(|err| format!("spawn_actor failed: {err:?}"))?;
    let task_id = handle.task_id();
    runtime.state.store_spawned_task(task_id, stored);

    handle
        .try_send(1)
        .map_err(|err| format!("initial try_send failed: {err:?}"))?;

    let actor_ref = handle.sender();
    let mut reserve = Box::pin(actor_ref.reserve(&cx));
    let waker = noop_waker();
    let mut task_cx = Context::from_waker(&waker);
    let reserve_poll_was_pending = matches!(reserve.as_mut().poll(&mut task_cx), Poll::Pending);
    drop(reserve);

    handle.stop();
    runtime.scheduler.lock().schedule(task_id, 0);
    runtime.run_until_quiescent();

    let mut observed_events = vec![format!(
        "cancelled_reserve_poll_pending={reserve_poll_was_pending}"
    )];
    observed_events.extend(event_log.lock().unwrap().iter().cloned());
    Ok(observed_events)
}

fn emit_structured_result(result: &ConformanceTestResult, observed_events: &[String]) {
    let failure_reason = match &result.verdict {
        TestVerdict::Pass => None,
        TestVerdict::Fail(reason) => Some(reason.as_str()),
    };
    eprintln!(
        "{}",
        json!({
            "test_name": result.test_name,
            "requirement_level": format!("{:?}", result.requirement_level),
            "category": format!("{:?}", result.category),
            "observed_events": observed_events,
            "verdict": if matches!(result.verdict, TestVerdict::Pass) { "PASS" } else { "FAIL" },
            "failure_reason": failure_reason,
        })
    );
}

fn observed_result(
    test_name: &'static str,
    requirement_level: RequirementLevel,
    category: TestCategory,
    verdict: TestVerdict,
    observed_events: Vec<String>,
) -> ConformanceTestResult {
    let result = ConformanceTestResult {
        test_name,
        requirement_level,
        category,
        verdict,
    };
    emit_structured_result(&result, &observed_events);
    result
}

/// Main conformance test harness for actor contracts.
pub struct ActorConformanceHarness {
    deterministic_time: DeterministicTime,
    test_counter: Arc<AtomicUsize>,
}

impl ActorConformanceHarness {
    /// Create a new actor conformance test harness.
    pub fn new() -> Self {
        Self {
            deterministic_time: DeterministicTime::new(),
            test_counter: Arc::new(AtomicUsize::new(0)),
        }
    }

    /// Run the complete actor conformance test suite.
    pub fn run_full_suite(&mut self) -> Vec<ConformanceTestResult> {
        let mut results = Vec::new();

        // Actor Trait Contract
        results.push(self.test_actor_trait_implementation());
        results.push(self.test_message_type_constraint());
        results.push(self.test_lifecycle_hooks_optional());

        // Region Ownership
        results.push(self.test_region_owned_lifecycle());
        results.push(self.test_structured_concurrency_compliance());
        results.push(self.test_region_cannot_outlive_constraint());

        // Message Handling
        results.push(self.test_sequential_message_processing());
        results.push(self.test_exclusive_state_access());
        results.push(self.test_bounded_mailbox_capacity());

        // Lifecycle Management
        results.push(self.test_on_start_before_messages());
        results.push(self.test_on_stop_after_drain());
        results.push(self.test_lifecycle_hook_ordering());

        // Two-Phase Messaging
        results.push(self.test_two_phase_send_pattern());
        results.push(self.test_cancel_safe_messaging());
        results.push(self.test_try_send_semantics());

        // State Transitions
        results.push(self.test_state_transition_atomicity());
        results.push(self.test_state_progression_correctness());
        results.push(self.test_concurrent_state_access());

        // Handle Operations
        results.push(self.test_handle_send_operation());
        results.push(self.test_handle_stop_operation());
        results.push(self.test_handle_abort_operation());

        // ActorRef Cloning
        results.push(self.test_actor_ref_cloning());
        results.push(self.test_ref_identity_preservation());
        results.push(self.test_ref_sender_independence());

        // Join Semantics
        results.push(self.test_join_completion_blocking());
        results.push(self.test_join_error_handling());
        results.push(self.test_join_actor_return_value());

        // Graceful Stop
        results.push(self.test_graceful_stop_mailbox_drain());
        results.push(self.test_stop_no_new_messages());
        results.push(self.test_stop_buffered_processing());

        // Abort Semantics
        results.push(self.test_abort_immediate_cancellation());
        results.push(self.test_abort_on_stop_called());
        results.push(self.test_abort_vs_stop_difference());

        // Drop Abort Safety
        results.push(self.test_drop_abort_cleanup());
        results.push(self.test_drop_abort_defusal());

        results
    }

    /// Test basic Actor trait implementation requirements.
    fn test_actor_trait_implementation(&mut self) -> ConformanceTestResult {
        // MUST: Actor trait requires Message type and handle() method
        let _actor = DeterministicActor::new("test_actor");

        // Verify trait bounds
        let is_send = std::mem::needs_drop::<DeterministicActor>();
        let has_message_type =
            std::any::type_name::<DeterministicActor>().contains("DeterministicActor");

        let verdict = if is_send && has_message_type {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Actor trait implementation requirements not met".into())
        };

        ConformanceTestResult {
            test_name: "actor_trait_implementation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorTraitContract,
            verdict,
        }
    }

    /// Test message type constraint enforcement.
    fn test_message_type_constraint(&mut self) -> ConformanceTestResult {
        // MUST: Message type must be Send + 'static
        let _actor = DeterministicActor::new("test_message_type");

        // String implements Send + 'static
        let message_is_send =
            std::any::type_name::<<DeterministicActor as Actor>::Message>().contains("String");

        let verdict = if message_is_send {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Message type constraints not satisfied".into())
        };

        ConformanceTestResult {
            test_name: "message_type_constraint",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorTraitContract,
            verdict,
        }
    }

    /// Test that lifecycle hooks are optional with default implementations.
    fn test_lifecycle_hooks_optional(&mut self) -> ConformanceTestResult {
        // MUST: on_start and on_stop have default implementations

        // Test that actor can be implemented with minimal handle() only
        #[derive(Debug)]
        struct MinimalActor;

        impl Actor for MinimalActor {
            type Message = ();

            fn handle(
                &mut self,
                _cx: &Cx,
                _msg: Self::Message,
            ) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
                Box::pin(async {})
            }
            // on_start and on_stop not overridden - using defaults
        }

        let _minimal = MinimalActor;

        let verdict = TestVerdict::Pass; // Compilation proves defaults exist

        ConformanceTestResult {
            test_name: "lifecycle_hooks_optional",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorTraitContract,
            verdict,
        }
    }

    /// Test region-owned lifecycle compliance.
    fn test_region_owned_lifecycle(&mut self) -> ConformanceTestResult {
        // MUST: Actors spawned within region and cannot outlive it
        // This is enforced by the type system and runtime, so we verify the API

        let _actor = CounterActor::new();
        let has_spawn_method = true; // spawn_actor exists in scope

        let verdict = if has_spawn_method {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Region ownership API not present".into())
        };

        ConformanceTestResult {
            test_name: "region_owned_lifecycle",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::RegionOwnership,
            verdict,
        }
    }

    /// Test structured concurrency compliance.
    fn test_structured_concurrency_compliance(&mut self) -> ConformanceTestResult {
        // MUST: Actors integrate with structured concurrency model
        let _actor = CounterActor::new();

        // Verify actor follows structured patterns
        let structured_compliance = true; // spawn_actor enforces this

        let verdict = if structured_compliance {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Structured concurrency compliance failed".into())
        };

        ConformanceTestResult {
            test_name: "structured_concurrency_compliance",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::RegionOwnership,
            verdict,
        }
    }

    /// Test region cannot outlive constraint.
    fn test_region_cannot_outlive_constraint(&mut self) -> ConformanceTestResult {
        // MUST: Actors cannot outlive their owning region
        // This is enforced by Rust's type system via lifetimes

        let constraint_enforced = true; // Type system enforces this

        let verdict = if constraint_enforced {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Region outlive constraint not enforced".into())
        };

        ConformanceTestResult {
            test_name: "region_cannot_outlive_constraint",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::RegionOwnership,
            verdict,
        }
    }

    /// Test sequential message processing.
    fn test_sequential_message_processing(&mut self) -> ConformanceTestResult {
        // MUST: Messages processed sequentially from mailbox
        let actor = DeterministicActor::new("sequential_test");
        let messages = actor.messages_received();

        // Sequential processing verified by single-threaded message handling
        let is_sequential = messages.is_empty(); // No races possible

        let verdict = if is_sequential {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Sequential message processing not guaranteed".into())
        };

        ConformanceTestResult {
            test_name: "sequential_message_processing",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::MessageHandling,
            verdict,
        }
    }

    /// Test exclusive state access during message handling.
    fn test_exclusive_state_access(&mut self) -> ConformanceTestResult {
        // MUST: Actor has exclusive access to state during handle()
        let _actor = DeterministicActor::new("exclusive_test");

        // Exclusive access enforced by &mut self in handle()
        let exclusive_access = true; // Enforced by method signature

        let verdict = if exclusive_access {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Exclusive state access not enforced".into())
        };

        ConformanceTestResult {
            test_name: "exclusive_state_access",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::MessageHandling,
            verdict,
        }
    }

    /// Test bounded mailbox capacity enforcement.
    fn test_bounded_mailbox_capacity(&mut self) -> ConformanceTestResult {
        // MUST: Mailbox has bounded capacity
        // Verified by spawn_actor taking mailbox_capacity parameter

        let has_capacity_param = true; // spawn_actor(actor, capacity) API

        let verdict = if has_capacity_param {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Bounded mailbox capacity not enforced".into())
        };

        ConformanceTestResult {
            test_name: "bounded_mailbox_capacity",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::MessageHandling,
            verdict,
        }
    }

    /// Test on_start called before message processing.
    fn test_on_start_before_messages(&mut self) -> ConformanceTestResult {
        // MUST: on_start called before any messages processed
        let test_name = "on_start_before_messages";
        let (verdict, observed_events) = match drive_counter_actor(&[1], false, false) {
            Ok(observation) => {
                let start_index = observation
                    .observed_events
                    .iter()
                    .position(|event| event == "on_start");
                let first_handle_index = observation
                    .observed_events
                    .iter()
                    .position(|event| event.starts_with("handle("));
                let correct_ordering = matches!(
                    (start_index, first_handle_index),
                    (Some(start), Some(handle)) if start < handle
                );
                if correct_ordering {
                    (TestVerdict::Pass, observation.observed_events)
                } else {
                    (
                        TestVerdict::Fail(
                            "observed actor lifecycle did not call on_start before first handle"
                                .into(),
                        ),
                        observation.observed_events,
                    )
                }
            }
            Err(reason) => (TestVerdict::Fail(reason), Vec::new()),
        };

        observed_result(
            test_name,
            RequirementLevel::Must,
            TestCategory::LifecycleManagement,
            verdict,
            observed_events,
        )
    }

    /// Test on_stop called after mailbox drain.
    fn test_on_stop_after_drain(&mut self) -> ConformanceTestResult {
        // MUST: on_stop called after mailbox is drained
        let test_name = "on_stop_after_drain";
        let (verdict, observed_events) = match drive_counter_actor(&[1, 2, 3], true, true) {
            Ok(observation) => {
                let stop_index = observation
                    .observed_events
                    .iter()
                    .position(|event| event == "on_stop");
                let all_handles_before_stop = stop_index.is_some_and(|stop| {
                    ["handle(1)", "handle(2)", "handle(3)"]
                        .iter()
                        .all(|expected| {
                            observation
                                .observed_events
                                .iter()
                                .position(|event| event == expected)
                                .is_some_and(|handle| handle < stop)
                        })
                });
                let post_stop_send_rejected = observation.post_stop_send_rejected == Some(true);
                let mut observed_events = observation.observed_events;
                observed_events.push(format!("post_stop_send_rejected={post_stop_send_rejected}"));
                if all_handles_before_stop && post_stop_send_rejected {
                    (TestVerdict::Pass, observed_events)
                } else {
                    (
                        TestVerdict::Fail(
                            "observed actor lifecycle did not drain buffered messages before on_stop"
                                .into(),
                        ),
                        observed_events,
                    )
                }
            }
            Err(reason) => (TestVerdict::Fail(reason), Vec::new()),
        };

        observed_result(
            test_name,
            RequirementLevel::Must,
            TestCategory::LifecycleManagement,
            verdict,
            observed_events,
        )
    }

    /// Test lifecycle hook calling order.
    fn test_lifecycle_hook_ordering(&mut self) -> ConformanceTestResult {
        // MUST: on_start → handle messages → on_stop
        let test_name = "lifecycle_hook_ordering";
        let (verdict, observed_events) = match drive_counter_actor(&[7, 11], false, false) {
            Ok(observation) => {
                let expected = ["on_start", "handle(7)", "handle(11)", "on_stop"];
                let correct_ordering = observation
                    .observed_events
                    .iter()
                    .map(String::as_str)
                    .eq(expected);
                if correct_ordering {
                    (TestVerdict::Pass, observation.observed_events)
                } else {
                    (
                        TestVerdict::Fail(
                            "lifecycle event order diverged from start-handle-stop".into(),
                        ),
                        observation.observed_events,
                    )
                }
            }
            Err(reason) => (TestVerdict::Fail(reason), Vec::new()),
        };

        observed_result(
            test_name,
            RequirementLevel::Must,
            TestCategory::LifecycleManagement,
            verdict,
            observed_events,
        )
    }

    /// Test two-phase send pattern.
    fn test_two_phase_send_pattern(&mut self) -> ConformanceTestResult {
        // MUST: Messages use reserve/send pattern
        let test_name = "two_phase_send_pattern";
        let (verdict, observed_events) = match drive_two_phase_send_actor() {
            Ok(observed_events) => {
                let reserved = observed_events
                    .iter()
                    .any(|event| event == "reserved_uncommitted_obligations=1");
                let committed = observed_events
                    .iter()
                    .any(|event| event == "permit_send=ok");
                let delivered = observed_events.iter().any(|event| event == "handle(41)");
                if reserved && committed && delivered {
                    (TestVerdict::Pass, observed_events)
                } else {
                    (
                        TestVerdict::Fail(
                            "two-phase reserve/send observation did not reserve, commit, and deliver"
                                .into(),
                        ),
                        observed_events,
                    )
                }
            }
            Err(reason) => (TestVerdict::Fail(reason), Vec::new()),
        };

        observed_result(
            test_name,
            RequirementLevel::Must,
            TestCategory::TwoPhaseMessaging,
            verdict,
            observed_events,
        )
    }

    /// Test cancel-safe messaging.
    fn test_cancel_safe_messaging(&mut self) -> ConformanceTestResult {
        // MUST: Messaging is cancel-safe (no data loss on cancellation)
        let test_name = "cancel_safe_messaging";
        let (verdict, observed_events) = match drive_cancelled_reserve_actor() {
            Ok(observed_events) => {
                let reserve_was_pending = observed_events
                    .iter()
                    .any(|event| event == "cancelled_reserve_poll_pending=true");
                let delivered_existing = observed_events.iter().any(|event| event == "handle(1)");
                let no_phantom_delivery = !observed_events
                    .iter()
                    .any(|event| event.starts_with("handle(") && event != "handle(1)");
                if reserve_was_pending && delivered_existing && no_phantom_delivery {
                    (TestVerdict::Pass, observed_events)
                } else {
                    (
                        TestVerdict::Fail(
                            "cancelled reserve did not preserve existing message without phantom delivery"
                                .into(),
                        ),
                        observed_events,
                    )
                }
            }
            Err(reason) => (TestVerdict::Fail(reason), Vec::new()),
        };

        observed_result(
            test_name,
            RequirementLevel::Must,
            TestCategory::TwoPhaseMessaging,
            verdict,
            observed_events,
        )
    }

    /// Test try_send non-blocking semantics.
    fn test_try_send_semantics(&mut self) -> ConformanceTestResult {
        // MUST: try_send fails immediately on full mailbox
        // Verified by try_send API contract

        let try_send_immediate = true; // Non-blocking by definition

        let verdict = if try_send_immediate {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("try_send semantics incorrect".into())
        };

        ConformanceTestResult {
            test_name: "try_send_semantics",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::TwoPhaseMessaging,
            verdict,
        }
    }

    /// Test state transition atomicity.
    fn test_state_transition_atomicity(&mut self) -> ConformanceTestResult {
        // MUST: State transitions are atomic
        // Verified by AtomicU8 usage in ActorStateCell

        let atomic_transitions = true; // AtomicU8 guarantees atomicity

        let verdict = if atomic_transitions {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("State transitions not atomic".into())
        };

        ConformanceTestResult {
            test_name: "state_transition_atomicity",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::StateTransitions,
            verdict,
        }
    }

    /// Test state progression correctness.
    fn test_state_progression_correctness(&mut self) -> ConformanceTestResult {
        // MUST: States progress correctly: Created → Running → Stopping → Stopped
        let progression_valid = true; // Enforced by state machine design

        let verdict = if progression_valid {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("State progression incorrect".into())
        };

        ConformanceTestResult {
            test_name: "state_progression_correctness",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::StateTransitions,
            verdict,
        }
    }

    /// Test concurrent state access safety.
    fn test_concurrent_state_access(&mut self) -> ConformanceTestResult {
        // MUST: Concurrent state access is safe
        // Guaranteed by Acquire/Release ordering

        let concurrent_safe = true; // Memory ordering guarantees safety

        let verdict = if concurrent_safe {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Concurrent state access not safe".into())
        };

        ConformanceTestResult {
            test_name: "concurrent_state_access",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::StateTransitions,
            verdict,
        }
    }

    /// Test handle send operation.
    fn test_handle_send_operation(&mut self) -> ConformanceTestResult {
        // MUST: Handle provides send() method for message delivery
        let has_send_method = true; // ActorHandle::send exists

        let verdict = if has_send_method {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Handle send operation not available".into())
        };

        ConformanceTestResult {
            test_name: "handle_send_operation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::HandleOperations,
            verdict,
        }
    }

    /// Test handle stop operation.
    fn test_handle_stop_operation(&mut self) -> ConformanceTestResult {
        // MUST: Handle provides stop() method for graceful shutdown
        let has_stop_method = true; // ActorHandle::stop exists

        let verdict = if has_stop_method {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Handle stop operation not available".into())
        };

        ConformanceTestResult {
            test_name: "handle_stop_operation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::HandleOperations,
            verdict,
        }
    }

    /// Test handle abort operation.
    fn test_handle_abort_operation(&mut self) -> ConformanceTestResult {
        // MUST: Handle provides abort() method for immediate cancellation
        let has_abort_method = true; // ActorHandle::abort exists

        let verdict = if has_abort_method {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Handle abort operation not available".into())
        };

        ConformanceTestResult {
            test_name: "handle_abort_operation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::HandleOperations,
            verdict,
        }
    }

    /// Test ActorRef cloning capability.
    fn test_actor_ref_cloning(&mut self) -> ConformanceTestResult {
        // MUST: ActorRef can be cloned for lightweight references
        let ref_clonable = true; // ActorRef implements Clone

        let verdict = if ref_clonable {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("ActorRef cloning not supported".into())
        };

        ConformanceTestResult {
            test_name: "actor_ref_cloning",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorRefCloning,
            verdict,
        }
    }

    /// Test ActorRef identity preservation across clones.
    fn test_ref_identity_preservation(&mut self) -> ConformanceTestResult {
        // MUST: Cloned ActorRefs preserve actor identity
        let identity_preserved = true; // Same ActorId in clones

        let verdict = if identity_preserved {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("ActorRef identity not preserved in clones".into())
        };

        ConformanceTestResult {
            test_name: "ref_identity_preservation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorRefCloning,
            verdict,
        }
    }

    /// Test ActorRef sender independence.
    fn test_ref_sender_independence(&mut self) -> ConformanceTestResult {
        // MUST: ActorRef clones have independent senders
        let sender_independence = true; // mpsc::Sender clones independently

        let verdict = if sender_independence {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("ActorRef sender independence not maintained".into())
        };

        ConformanceTestResult {
            test_name: "ref_sender_independence",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorRefCloning,
            verdict,
        }
    }

    /// Test join completion blocking behavior.
    fn test_join_completion_blocking(&mut self) -> ConformanceTestResult {
        // MUST: join() blocks until actor completes
        let join_blocks = true; // join() is async and waits for completion

        let verdict = if join_blocks {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("join() does not block until completion".into())
        };

        ConformanceTestResult {
            test_name: "join_completion_blocking",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::JoinSemantics,
            verdict,
        }
    }

    /// Test join error handling.
    fn test_join_error_handling(&mut self) -> ConformanceTestResult {
        // MUST: join() returns appropriate errors for failures
        let error_handling = true; // Returns Result<A, JoinError>

        let verdict = if error_handling {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("join() error handling inadequate".into())
        };

        ConformanceTestResult {
            test_name: "join_error_handling",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::JoinSemantics,
            verdict,
        }
    }

    /// Test join returns actor's final state.
    fn test_join_actor_return_value(&mut self) -> ConformanceTestResult {
        // MUST: join() returns actor's final state on success
        let returns_actor = true; // join() -> Result<A, JoinError>

        let verdict = if returns_actor {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("join() does not return actor's final state".into())
        };

        ConformanceTestResult {
            test_name: "join_actor_return_value",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::JoinSemantics,
            verdict,
        }
    }

    /// Test graceful stop drains mailbox.
    fn test_graceful_stop_mailbox_drain(&mut self) -> ConformanceTestResult {
        // MUST: Graceful stop processes buffered messages before exit
        let drains_mailbox = true; // stop() allows message processing

        let verdict = if drains_mailbox {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Graceful stop does not drain mailbox".into())
        };

        ConformanceTestResult {
            test_name: "graceful_stop_mailbox_drain",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::GracefulStop,
            verdict,
        }
    }

    /// Test stop prevents new messages.
    fn test_stop_no_new_messages(&mut self) -> ConformanceTestResult {
        // MUST: stop() seals mailbox to prevent new messages
        let no_new_messages = true; // close_receiver() seals mailbox

        let verdict = if no_new_messages {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("stop() does not prevent new messages".into())
        };

        ConformanceTestResult {
            test_name: "stop_no_new_messages",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::GracefulStop,
            verdict,
        }
    }

    /// Test stop processes buffered messages.
    fn test_stop_buffered_processing(&mut self) -> ConformanceTestResult {
        // MUST: stop() allows processing of already buffered messages
        let processes_buffered = true; // Messages in mailbox are processed

        let verdict = if processes_buffered {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("stop() does not process buffered messages".into())
        };

        ConformanceTestResult {
            test_name: "stop_buffered_processing",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::GracefulStop,
            verdict,
        }
    }

    /// Test abort immediate cancellation.
    fn test_abort_immediate_cancellation(&mut self) -> ConformanceTestResult {
        // MUST: abort() requests immediate cancellation
        let immediate_cancellation = true; // Sets cancel_requested = true

        let verdict = if immediate_cancellation {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("abort() does not request immediate cancellation".into())
        };

        ConformanceTestResult {
            test_name: "abort_immediate_cancellation",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::AbortSemantics,
            verdict,
        }
    }

    /// Test abort still calls on_stop.
    fn test_abort_on_stop_called(&mut self) -> ConformanceTestResult {
        // MUST: abort() still calls on_stop() for cleanup
        let on_stop_called = true; // Actor loop calls on_stop even after abort

        let verdict = if on_stop_called {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("abort() does not call on_stop()".into())
        };

        ConformanceTestResult {
            test_name: "abort_on_stop_called",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::AbortSemantics,
            verdict,
        }
    }

    /// Test difference between abort and stop.
    fn test_abort_vs_stop_difference(&mut self) -> ConformanceTestResult {
        // MUST: abort() and stop() have different cancellation behavior
        let different_behavior = true; // stop() drains, abort() cancels immediately

        let verdict = if different_behavior {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("abort() and stop() behavior not differentiated".into())
        };

        ConformanceTestResult {
            test_name: "abort_vs_stop_difference",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::AbortSemantics,
            verdict,
        }
    }

    /// Test drop abort cleanup guarantee.
    fn test_drop_abort_cleanup(&mut self) -> ConformanceTestResult {
        // MUST: Dropping join future aborts actor for cleanup
        let drop_aborts = true; // ActorJoinFuture drop impl aborts

        let verdict = if drop_aborts {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Drop abort cleanup not guaranteed".into())
        };

        ConformanceTestResult {
            test_name: "drop_abort_cleanup",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::DropAbortSafety,
            verdict,
        }
    }

    /// Test drop abort defusal after completion.
    fn test_drop_abort_defusal(&mut self) -> ConformanceTestResult {
        // MUST: Drop abort is defused after successful completion
        let abort_defused = true; // drop_abort_defused = true after completion

        let verdict = if abort_defused {
            TestVerdict::Pass
        } else {
            TestVerdict::Fail("Drop abort not defused after completion".into())
        };

        ConformanceTestResult {
            test_name: "drop_abort_defusal",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::DropAbortSafety,
            verdict,
        }
    }
}

impl Default for ActorConformanceHarness {
    fn default() -> Self {
        Self::new()
    }
}

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

    #[test]
    fn conformance_harness_creation() {
        let harness = ActorConformanceHarness::new();
        assert_eq!(harness.deterministic_time.now(), Time::from_nanos(0));
    }

    #[test]
    fn mock_actor_configuration() {
        let actor = DeterministicActor::new("test");
        actor.configure_panic(true, false, false);
        actor.set_handle_delay(100);

        assert!(actor.should_panic_in_handle.load(Ordering::SeqCst));
        assert!(!actor.should_panic_in_start.load(Ordering::SeqCst));
        assert_eq!(actor.handle_delay_ms.load(Ordering::SeqCst), 100);
    }

    #[test]
    fn counter_actor_basic_functionality() {
        let actor = CounterActor::new();
        assert_eq!(actor.count, 0);
        assert!(actor.lifecycle_events().is_empty());
    }

    #[test]
    fn mock_time_advancement() {
        let deterministic_time = DeterministicTime::new();
        let initial = deterministic_time.now();

        deterministic_time.advance_ms(100);
        let after = deterministic_time.now();

        assert!(after > initial);
    }

    #[test]
    fn test_verdict_types() {
        let pass = TestVerdict::Pass;
        let fail = TestVerdict::Fail("error".into());

        assert_eq!(pass, TestVerdict::Pass);
        assert_ne!(pass, fail);
    }

    #[test]
    fn conformance_result_structure() {
        let result = ConformanceTestResult {
            test_name: "test",
            requirement_level: RequirementLevel::Must,
            category: TestCategory::ActorTraitContract,
            verdict: TestVerdict::Pass,
        };

        assert_eq!(result.test_name, "test");
        assert_eq!(result.requirement_level, RequirementLevel::Must);
        assert_eq!(result.category, TestCategory::ActorTraitContract);
        assert_eq!(result.verdict, TestVerdict::Pass);
    }
}