//! # Actor Runtime
//!
//! The Actor Runtime is responsible for initializing, running, and managing the lifecycle
//! of WebAssembly actors within the Theater system. It coordinates the various components
//! that an actor needs to function, including execution, handlers, and communication channels.

use crate::actor::handle::ActorHandle;
use crate::actor::store::ActorStore;
use crate::actor::types::ActorError;
use crate::actor::types::ActorOperation;
use crate::events::wasm::WasmEventData;
use crate::events::ChainEventData;
use crate::handler::HandlerContext;
use crate::handler::HandlerRegistry;
use crate::id::TheaterId;
use crate::interceptor::{RecordingInterceptor, ReplayRecordingInterceptor};
use crate::messages::TheaterCommand;
use crate::metrics::MetricsCollector;
use crate::pack_bridge::{AsyncRuntime, CallInterceptor, HostLinkerBuilder, PackInstance, Value};

use crate::Result;
use crate::ShutdownController;
use crate::ShutdownType;
use crate::StateChain;
use std::sync::Arc;
use std::sync::RwLock as SyncRwLock;
use tokio::sync::mpsc::{self, Receiver, Sender};
use tokio::sync::RwLock;
use tokio::task::JoinHandle;
use tokio::time::Instant;
use tracing::{debug, error, info, warn};

use super::types::ActorControl;
use super::types::ActorInfo;

/// # ActorRuntime
///
/// Coordinates the execution and lifecycle of a single WebAssembly actor within the Theater system.
///
/// `ActorRuntime` provides static methods for building and running actors. It manages the various
/// components that make up an actor's execution environment, including handlers and communication
/// channels. It's responsible for starting the actor, setting up its capabilities via handlers,
/// executing operations, and ensuring proper shutdown.
pub struct ActorRuntime;

/// # Result of starting an actor
///
/// Represents the outcome of attempting to start an actor.
///
/// This enum provides detailed information about whether an actor was successfully
/// started or encountered errors during initialization. It includes the actor's ID
/// in both success and failure cases, and detailed error information in the failure case.
#[derive(Debug)]
pub enum StartActorResult {
    /// Actor successfully started
    Success(TheaterId),
    /// Actor failed to start with error message
    Failure(TheaterId, String),
    /// Actor failed to start with permission or validation error
    Error(String),
}

#[derive(Debug, thiserror::Error)]
pub enum ActorRuntimeError {
    #[error("Setup error: {message}")]
    SetupError { message: String },

    #[error("Actor instance not found: {message}")]
    ActorInstanceNotFound { message: String },

    #[error("Actor phase error: expected {expected:?}, found {found:?}. {message}")]
    ActorPhaseError {
        expected: ActorPhase,
        found: ActorPhase,
        message: String,
    },

    #[error("Actor error: {0}")]
    ActorError(#[from] ActorError),

    #[error(transparent)]
    UnknownError(#[from] anyhow::Error),
}

#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum ActorPhase {
    #[default]
    Starting,
    Running,
    Paused,
    ShuttingDown,
}

impl std::fmt::Display for ActorPhase {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ActorPhase::Starting => write!(f, "Starting"),
            ActorPhase::Running => write!(f, "Running"),
            ActorPhase::Paused => write!(f, "Paused"),
            ActorPhase::ShuttingDown => write!(f, "Shutting Down"),
        }
    }
}

#[derive(Clone)]
pub struct ActorPhaseManager {
    phase_tx: Arc<tokio::sync::watch::Sender<ActorPhase>>,
    phase_rx: tokio::sync::watch::Receiver<ActorPhase>,
}

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

impl ActorPhaseManager {
    pub fn new() -> Self {
        let (tx, rx) = tokio::sync::watch::channel(ActorPhase::Starting);
        Self {
            phase_tx: Arc::new(tx),
            phase_rx: rx,
        }
    }

    pub fn set_phase(&self, phase: ActorPhase) {
        let _ = self.phase_tx.send(phase);
    }

    pub fn get_phase(&self) -> ActorPhase {
        self.phase_rx.borrow().clone()
    }

    pub fn is_phase(&self, phase: ActorPhase) -> bool {
        *self.phase_rx.borrow() == phase
    }

    pub async fn wait_for_phase(&self, phase: ActorPhase) {
        let mut rx = self.phase_rx.clone();
        let _ = rx.wait_for(|current_phase| *current_phase == phase).await;
    }

    /// Wait until the actor reaches Running or ShuttingDown.
    /// Returns true if Running, false if ShuttingDown.
    pub async fn wait_for_ready_or_shutdown(&self) -> bool {
        let mut rx = self.phase_rx.clone();
        let _ = rx
            .wait_for(|current_phase| {
                *current_phase == ActorPhase::Running || *current_phase == ActorPhase::ShuttingDown
            })
            .await;
        self.is_phase(ActorPhase::Running)
    }
}

impl ActorRuntime {
    #[allow(clippy::too_many_arguments)]
    pub async fn build_actor_resources(
        id: TheaterId,
        name: String,
        wasm_bytes: Vec<u8>,
        pack_runtime: AsyncRuntime,
        chain: Arc<SyncRwLock<StateChain>>,
        handler_registry: HandlerRegistry,
        theater_tx: Sender<TheaterCommand>,
        operation_tx: Sender<ActorOperation>,
        info_tx: Sender<ActorInfo>,
        control_tx: Sender<ActorControl>,
        actor_phase_manager: ActorPhaseManager,
        actor_instance_wrapper: Arc<RwLock<Option<PackInstance>>>,
        initial_state: Value,
    ) -> Result<(ShutdownController, Vec<JoinHandle<()>>), ActorRuntimeError> {
        // ---------------- Checkpoint Setup Initial ----------------

        debug!("Setting up actor store");

        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Setup Initial".into(),
            });
        }

        let handle_operation_tx = operation_tx.clone();
        let actor_handle = ActorHandle::new(handle_operation_tx, info_tx, control_tx);
        let actor_store = ActorStore::new(
            id,
            theater_tx.clone(),
            actor_handle.clone(),
            chain.clone(),
            initial_state,
        );

        // ----------------- Checkpoint Store Manifest ----------------

        debug!("Storing manifest for actor: {}", id);

        // Checkpoint 1: After manifest storage
        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Store Manifest".into(),
            });
        }

        // ----------------- Checkpoint Load Component -----------------

        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Load Package".into(),
            });
        }

        debug!(
            "Using pre-resolved WASM bytes for actor {} ({} bytes)",
            name,
            wasm_bytes.len()
        );

        // WASM bytes are now passed in directly from spawn_actor()
        // This allows callers to provide pre-compiled WASM or for the
        // theater runtime to resolve the package reference early.

        // ----------------- Checkpoint Get Handlers -----------------

        debug!("Getting handlers from registry");

        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Get Handlers".into(),
            });
        }

        // Get all handlers from the registry
        let mut handlers = handler_registry.get_handlers();
        debug!("Got {} handlers from registry", handlers.len());

        // ----------------- Checkpoint Instantiate with Host Functions -----------------

        debug!("Creating PackInstance with host functions");

        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Instantiate".into(),
            });
        }

        // Create the appropriate interceptor based on mode
        let interceptor: Option<Arc<dyn CallInterceptor>> = if handler_registry.is_replay_mode() {
            // In replay mode, use the expected chain from the registry to look up
            // recorded host function outputs, and record new events to the actor's chain.
            let expected_events = handler_registry
                .replay_chain()
                .expect("Replay mode set but no replay chain provided")
                .clone();
            Some(Arc::new(ReplayRecordingInterceptor::new(
                expected_events,
                chain.clone(),
            )))
        } else {
            Some(Arc::new(RecordingInterceptor::new(chain.clone())))
        };

        // Create shutdown controller early so handlers can subscribe during setup
        let mut shutdown_controller = ShutdownController::new();

        // Create handler context with shutdown controller access
        let mut handler_ctx = HandlerContext::with_shutdown_controller(shutdown_controller.clone());
        handler_ctx.actor_id = Some(id);
        let handlers_for_setup = &mut handlers;

        let mut actor_instance = PackInstance::new_with_interceptor(
            name.clone(),
            &wasm_bytes,
            &pack_runtime,
            actor_store,
            interceptor,
            |builder: &mut HostLinkerBuilder<'_, ActorStore>| {
                // Set up host functions for each handler
                for handler in handlers_for_setup.iter_mut() {
                    debug!(
                        "Setting up Composite host functions for handler '{}'",
                        handler.name()
                    );
                    match handler.setup_host_functions_composite(builder, &mut handler_ctx) {
                        Ok(()) => {
                            debug!(
                                "Handler '{}' Composite host functions set up successfully",
                                handler.name()
                            );
                        }
                        Err(e) => {
                            error!(
                                "Handler '{}' Composite host functions FAILED: {:?}",
                                handler.name(),
                                e
                            );
                            return Err(e);
                        }
                    }
                }
                Ok(())
            },
        )
        .await
        .map_err(|e| {
            let error_message = format!("Failed to instantiate actor {}: {}", id, e);
            error!("{}", error_message);
            ActorRuntimeError::SetupError {
                message: error_message,
            }
        })?;

        debug!("PackInstance created successfully");
        debug!(
            "Handler context satisfied imports: {:?}",
            handler_ctx.satisfied_imports
        );

        // Verify interface hash compatibility between actor imports and handler interfaces.
        // This provides O(1) structural compatibility checking using Merkle-tree hashes.
        // Hash mismatches are FATAL - the actor will not start if signatures don't match.
        //
        // PARTIAL INTERFACE MATCHING: An actor may import only a subset of functions
        // from an interface. For example, an actor might import only `log` from
        // `theater:simple/runtime` which provides `log`, `get-chain`, and `shutdown`.
        // In this case, we compute a subset hash from the handler's functions and
        // compare against the actor's declared interface hash.
        match actor_instance.get_metadata_with_hashes().await {
            Ok(metadata) => {
                let actor_import_hashes = &metadata.import_hashes;
                info!(
                    "Verifying interface hashes for actor {} ({} imports)",
                    id,
                    actor_import_hashes.len()
                );

                for actor_import in actor_import_hashes {
                    let interface_name = &actor_import.name;
                    let actor_hash = &actor_import.hash;

                    // Get the function names the actor imports from this interface
                    let actor_function_names =
                        metadata.arena.imported_function_names(interface_name);
                    let function_name_refs: Vec<&str> =
                        actor_function_names.iter().map(|s| s.as_str()).collect();

                    // Find a handler that provides this interface and satisfies all required functions
                    let mut interface_verified = false;
                    let mut last_error: Option<String> = None;

                    for handler in handlers.iter() {
                        // Get the InterfaceImpl objects from the handler
                        let handler_interfaces = handler.interfaces();
                        if let Some(handler_interface) = handler_interfaces
                            .iter()
                            .find(|i| i.name() == interface_name)
                        {
                            // Compute the handler's subset hash for just those functions
                            match handler_interface.hash_subset(&function_name_refs) {
                                Some(subset_hash) => {
                                    if *actor_hash == subset_hash {
                                        info!(
                                            "Interface '{}' hash verified via subset matching (handler: {}, functions: {:?})",
                                            interface_name,
                                            handler.name(),
                                            actor_function_names
                                        );
                                        interface_verified = true;
                                        break;
                                    } else {
                                        // Check if full interface matches (shouldn't happen, but check)
                                        if *actor_hash == handler_interface.hash() {
                                            info!(
                                                "Interface '{}' hash verified (full interface, handler: {})",
                                                interface_name,
                                                handler.name()
                                            );
                                            interface_verified = true;
                                            break;
                                        }

                                        // Hash mismatch - record error but try other handlers
                                        last_error = Some(format!(
                                            "Interface hash mismatch for '{}' (handler: {}). \
                                            The actor's function signatures don't match the handler's implementation. \
                                            Actor imports functions: {:?}. \
                                            Actor hash: {}, Handler subset hash: {}",
                                            interface_name,
                                            handler.name(),
                                            actor_function_names,
                                            hex::encode(actor_hash.as_bytes()),
                                            hex::encode(subset_hash.as_bytes())
                                        ));
                                    }
                                }
                                None => {
                                    // Handler is missing some functions - try other handlers
                                    debug!(
                                        "Handler '{}' is missing function(s) for interface '{}'. \
                                        Actor imports: {:?}. Checking other handlers.",
                                        handler.name(),
                                        interface_name,
                                        actor_function_names
                                    );
                                }
                            }
                        }
                    }

                    if !interface_verified {
                        // No handler could satisfy this interface
                        let error_msg = last_error.unwrap_or_else(|| {
                            format!(
                                "No handler provides interface '{}' required by actor {}",
                                interface_name, id
                            )
                        });
                        error!("{}", error_msg);
                        return Err(ActorRuntimeError::SetupError { message: error_msg });
                    }
                }

                info!("All interface hashes verified for actor {}", id);

                // Cache function type info for host-side validation.
                // This is mandatory — we guarantee type safety to actors.
                actor_instance.cache_function_types().await.map_err(|e| {
                    ActorRuntimeError::SetupError {
                        message: format!(
                            "Failed to cache function types for actor {}: {:?}. \
                             All actors must have valid type metadata.",
                            id, e
                        ),
                    }
                })?;
            }
            Err(e) => {
                // Actor doesn't have __pack_types metadata - FATAL
                // All Pack actors MUST export __pack_types with interface hashes
                let error_msg = format!(
                    "Actor {} has no interface metadata (__pack_types). \
                    All actors must embed type metadata for interface verification. Error: {:?}",
                    id, e
                );
                error!("{}", error_msg);
                return Err(ActorRuntimeError::SetupError { message: error_msg });
            }
        }

        // Note: Export discovery is now automatic via Pack's embedded __pack_types metadata.
        // No manual export registration is needed.

        // ----------------- Checkpoint Finalize Setup -----------------

        debug!("Ready");

        if !actor_phase_manager.is_phase(ActorPhase::Starting) {
            let curr_phase = actor_phase_manager.get_phase();
            return Err(ActorRuntimeError::ActorPhaseError {
                expected: ActorPhase::Starting,
                found: curr_phase,
                message: "phase error found at setup task Checkpoint Finalize Setup".into(),
            });
        }

        // Put actor_instance in the shared wrapper
        {
            let mut instance_guard = actor_instance_wrapper.write().await;
            *instance_guard = Some(actor_instance);
        }

        // Set up handlers - use the shutdown_controller created earlier
        let mut handler_tasks: Vec<JoinHandle<()>> = vec![];
        let handler_actor_handle = actor_handle.clone();
        debug!("Setting up {} handlers", handlers.len());
        for mut handler in handlers {
            let handler_name = handler.name().to_string();
            debug!("Spawning task for handler: {}", handler_name);
            let actor_handle = handler_actor_handle.clone();
            let actor_instance = actor_instance_wrapper.clone();
            let shutdown_receiver = shutdown_controller.subscribe();
            // Create event subscription for streaming verification
            let event_rx = {
                let chain_guard = chain.read().unwrap();
                chain_guard.subscribe()
            };
            let handler_task = tokio::spawn(async move {
                debug!("Handler task running: {}", handler_name);
                if let Err(e) = handler
                    .setup(actor_handle, actor_instance, shutdown_receiver, event_rx)
                    .await
                {
                    error!("Handler '{}' setup() failed: {:?}", handler_name, e);
                } else {
                    debug!("Handler '{}' setup() completed", handler_name);
                }
            });
            handler_tasks.push(handler_task);
        }

        Ok((shutdown_controller, handler_tasks))
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn start(
        id: TheaterId,
        name: String,
        wasm_bytes: Vec<u8>,
        pack_runtime: AsyncRuntime,
        chain: Arc<SyncRwLock<StateChain>>,
        handler_registry: HandlerRegistry,
        theater_tx: Sender<TheaterCommand>,
        operation_rx: Receiver<ActorOperation>,
        operation_tx: Sender<ActorOperation>,
        info_rx: Receiver<ActorInfo>,
        info_tx: Sender<ActorInfo>,
        control_rx: Receiver<ActorControl>,
        control_tx: Sender<ActorControl>,
        initial_state: Value,
        setup_result_tx: Option<tokio::sync::oneshot::Sender<Result<(), String>>>,
    ) {
        info!("Actor runtime starting communication loops");
        let actor_phase_manager = ActorPhaseManager::new();

        // These will be set once setup completes
        let actor_instance_wrapper: Arc<RwLock<Option<PackInstance>>> = Arc::new(RwLock::new(None));
        let metrics: Arc<RwLock<MetricsCollector>> = Arc::new(RwLock::new(MetricsCollector::new()));
        let handler_tasks: Arc<RwLock<Vec<JoinHandle<()>>>> = Arc::new(RwLock::new(vec![]));
        let handlers_shutdown_controller: Arc<RwLock<Option<ShutdownController>>> =
            Arc::new(RwLock::new(None));
        let operation_rx = operation_rx;
        let info_rx = info_rx;
        let mut control_rx = control_rx;

        let mut setup_handle = {
            let actor_instance_wrapper = actor_instance_wrapper.clone();
            let actor_phase_manager = actor_phase_manager.clone();
            let name = name.clone();
            let theater_tx = theater_tx.clone();
            let handler_tasks = handler_tasks.clone();
            let handlers_shutdown_controller = handlers_shutdown_controller.clone();

            tokio::spawn(async move {
                match Self::build_actor_resources(
                    id,
                    name,
                    wasm_bytes,
                    pack_runtime,
                    chain,
                    handler_registry,
                    theater_tx,
                    operation_tx,
                    info_tx,
                    control_tx,
                    actor_phase_manager.clone(),
                    actor_instance_wrapper,
                    initial_state,
                )
                .await
                {
                    Ok((shutdown_controller, handlers)) => {
                        {
                            let mut handler_tasks_guard = handler_tasks.write().await;
                            *handler_tasks_guard = handlers;
                        }
                        {
                            let mut shutdown_controller_guard =
                                handlers_shutdown_controller.write().await;
                            *shutdown_controller_guard = Some(shutdown_controller);
                        }

                        actor_phase_manager.set_phase(ActorPhase::Running);
                        info!("Actor setup complete, now running");

                        // Signal success to spawn_actor
                        if let Some(tx) = setup_result_tx {
                            let _ = tx.send(Ok(()));
                        }
                    }
                    Err(e) => {
                        let error_msg = format!("{}", e);
                        error!("Failed to set up actor runtime: {}", error_msg);
                        // Set phase to ShuttingDown so operation_loop and other loops can exit
                        actor_phase_manager.set_phase(ActorPhase::ShuttingDown);

                        // Signal failure to spawn_actor
                        if let Some(tx) = setup_result_tx {
                            let _ = tx.send(Err(error_msg));
                        }
                    }
                }
            })
        };

        let mut info_handle = {
            let actor_instance_wrapper = actor_instance_wrapper.clone();
            let metrics = metrics.clone();
            let actor_phase_manager = actor_phase_manager.clone();

            tokio::spawn(Self::info_loop(
                info_rx,
                actor_instance_wrapper,
                metrics,
                actor_phase_manager,
            ))
        };

        let mut operation_handle = {
            let actor_instance_wrapper = actor_instance_wrapper.clone();
            let metrics = metrics.clone();
            let theater_tx = theater_tx.clone();
            let actor_phase_manager = actor_phase_manager.clone();

            tokio::spawn(Self::operation_loop(
                operation_rx,
                actor_instance_wrapper,
                metrics,
                theater_tx,
                actor_phase_manager,
            ))
        };

        while let Some(control) = control_rx.recv().await {
            info!("Received control command: {:?}", control);
            match control {
                ActorControl::Shutdown { response_tx } => {
                    info!("Shutdown requested");
                    actor_phase_manager.set_phase(ActorPhase::ShuttingDown);

                    // Signal handlers FIRST so they can cancel any blocking operations
                    // (e.g., TCP receive() calls that are waiting for data)
                    match handlers_shutdown_controller.write().await.take() {
                        Some(controller) => {
                            controller.signal_shutdown(ShutdownType::Graceful).await;
                        }
                        None => {
                            warn!("No handlers shutdown controller found");
                        }
                    }
                    info!("Signaled shutdown to handlers");

                    info!("Waiting for operation and info loops to finish");

                    // Wait for each individually to identify which one hangs
                    let _op_result = tokio::select! {
                        r = &mut operation_handle => {
                            info!("Operation loop exited: {:?}", r.is_ok());
                            r
                        }
                        _ = tokio::time::sleep(std::time::Duration::from_secs(5)) => {
                            error!("Operation loop did not exit within 5s - aborting");
                            operation_handle.abort();
                            operation_handle.await
                        }
                    };

                    let _info_result = tokio::select! {
                        r = &mut info_handle => {
                            info!("Info loop exited: {:?}", r.is_ok());
                            r
                        }
                        _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {
                            error!("Info loop did not exit within 2s - aborting");
                            info_handle.abort();
                            info_handle.await
                        }
                    };

                    let _setup_result = tokio::select! {
                        r = &mut setup_handle => {
                            info!("Setup loop exited: {:?}", r.is_ok());
                            r
                        }
                        _ = tokio::time::sleep(std::time::Duration::from_secs(2)) => {
                            error!("Setup loop did not exit within 2s - aborting");
                            setup_handle.abort();
                            setup_handle.await
                        }
                    };

                    info!("All loops have exited");

                    if let Err(e) = response_tx.send(Ok(())) {
                        error!("Failed to send shutdown confirmation: {:?}", e);
                    }

                    debug!("Shutdown confirmation sent, exiting control loop");
                    break;
                }
                ActorControl::Terminate { response_tx } => {
                    info!("Terminate requested");
                    // Abort info and operation loops
                    operation_handle.abort();
                    info_handle.abort();
                    setup_handle.abort();
                    match handlers_shutdown_controller.write().await.take() {
                        Some(controller) => {
                            controller.signal_shutdown(ShutdownType::Force).await;
                        }
                        None => {
                            warn!("No handlers shutdown controller found");
                        }
                    }
                    if let Err(e) = response_tx.send(Ok(())) {
                        error!("Failed to send terminate confirmation: {:?}", e);
                    }
                    break;
                }
                ActorControl::Pause { response_tx } => {
                    if actor_phase_manager.is_phase(ActorPhase::ShuttingDown) {
                        let _ = response_tx.send(Err(ActorError::ShuttingDown));
                    } else {
                        actor_phase_manager.set_phase(ActorPhase::Paused);
                        let _ = response_tx.send(Ok(()));
                    }
                }
                ActorControl::Resume { response_tx } => match actor_phase_manager.get_phase() {
                    ActorPhase::Starting | ActorPhase::Running => {
                        let _ = response_tx.send(Err(ActorError::NotPaused));
                    }
                    ActorPhase::ShuttingDown => {
                        let _ = response_tx.send(Err(ActorError::ShuttingDown));
                    }
                    ActorPhase::Paused => {
                        actor_phase_manager.set_phase(ActorPhase::Running);
                        let _ = response_tx.send(Ok(()));
                    }
                },
            }
        }

        // Gonna have to send the shutdown signal to all our handlers / respond to the shutdown
        // request

        info!("Actor runtime communication loop exiting, performing cleanup");
        let metrics = metrics.read().await;

        // If any handlers are still running, abort them
        let handler_tasks = handler_tasks.read().await;
        for handle in handler_tasks.iter() {
            if !handle.is_finished() {
                info!("Aborting handler task");
                handle.abort();
            }
        }

        // Log final metrics
        let final_metrics = metrics.get_metrics().await;
        info!("Final metrics at shutdown: {:?}", final_metrics);

        info!("Actor runtime cleanup complete");
    }

    async fn operation_loop(
        mut operation_rx: Receiver<ActorOperation>,
        actor_instance_wrapper: Arc<RwLock<Option<PackInstance>>>,
        metrics: Arc<RwLock<MetricsCollector>>,
        theater_tx: Sender<TheaterCommand>,
        actor_phase_manager: ActorPhaseManager,
    ) {
        // Wait for actor to be ready or fail during setup
        if !actor_phase_manager.wait_for_ready_or_shutdown().await {
            // Actor setup failed - exit immediately
            debug!("Actor setup failed, operation_loop exiting");
            return;
        }

        loop {
            tokio::select! {
                biased;

                _ = actor_phase_manager.wait_for_phase(ActorPhase::ShuttingDown) => {
                    break;
                }

                _ = actor_phase_manager.wait_for_phase(ActorPhase::Paused) => {
                    actor_phase_manager.wait_for_phase(ActorPhase::Running).await;
                }

                Some(op) = operation_rx.recv() => {
                    Self::process_operation(
                        op, &actor_instance_wrapper, &metrics, &theater_tx, actor_phase_manager.clone()
                    ).await
                }

                else => break,
            }
        }
    }

    async fn process_operation(
        op: ActorOperation,
        actor_instance_wrapper: &Arc<RwLock<Option<PackInstance>>>,
        metrics: &Arc<RwLock<MetricsCollector>>,
        theater_tx: &Sender<TheaterCommand>,
        actor_phase_manager: ActorPhaseManager,
    ) {
        match op {
            ActorOperation::CallFunctionPack {
                name,
                params,
                response_tx,
            } => {
                info!("Processing pack function call: {}", name);
                let mut actor_instance_guard = actor_instance_wrapper.write().await;
                let actor_instance = match &mut *actor_instance_guard {
                    Some(instance) => instance,
                    None => {
                        let err = ActorRuntimeError::ActorInstanceNotFound {
                            message: "Actor instance not found".to_string(),
                        };

                        let _ = theater_tx
                            .send(TheaterCommand::ActorRuntimeError { error: err })
                            .await;

                        let actor_err =
                            ActorError::UnexpectedError("Actor instance not found".to_string());

                        if let Err(e) = response_tx.send(Err(actor_err)) {
                            error!(
                                "Failed to send function call error response for operation '{}': {:?}",
                                name, e
                            );
                        }
                        return;
                    }
                };
                let metrics = metrics.write().await;
                match Self::execute_call_pack(actor_instance, &name, params, theater_tx, &metrics)
                    .await
                {
                    Ok(result) => {
                        if let Err(e) = response_tx.send(Ok(result)) {
                            error!(
                                "Failed to send function call response for operation '{}': {:?}",
                                name, e
                            );
                        }
                    }
                    Err(actor_error) => {
                        let _ = theater_tx
                            .send(TheaterCommand::ActorError {
                                actor_id: actor_instance.id(),
                                error: actor_error.clone(),
                            })
                            .await;

                        error!("Operation '{}' failed with error: {:?}", name, actor_error);
                        if let Err(send_err) = response_tx.send(Err(actor_error)) {
                            error!("Failed to send function call error response for operation '{}': {:?}", name, send_err);
                        }

                        // Pause the actor on error
                        actor_phase_manager.set_phase(ActorPhase::Paused);
                    }
                }
            }
            ActorOperation::HandleWasiHttpRequest { response_tx, .. } => {
                // WASI HTTP incoming requests are handled directly by the http handler
                // via the SharedActorInstance, not through this operation channel.
                // If this operation is received, it indicates a configuration error.
                let err = ActorError::UnexpectedError(
                    "HandleWasiHttpRequest should be handled by the HTTP handler directly"
                        .to_string(),
                );
                let _ = response_tx.send(Err(err));
            }
        }
    }

    async fn info_loop(
        mut info_rx: Receiver<ActorInfo>,
        actor_instance_wrapper: Arc<RwLock<Option<PackInstance>>>,
        metrics: Arc<RwLock<MetricsCollector>>,
        actor_phase_manager: ActorPhaseManager,
    ) {
        // Handle info requests
        loop {
            tokio::select! {
                    biased;

                    _ = actor_phase_manager.wait_for_phase(ActorPhase::ShuttingDown) => {
                        break;
                    }

                    Some(info) = info_rx.recv() => {
                        info!("Received info request: {:?}", info);
                        match info {
                    ActorInfo::GetStatus { response_tx } => {
                        let status = actor_phase_manager.get_phase().to_string();

                        if let Err(e) = response_tx.send(Ok(status)) {
                            error!("Failed to send status response: {:?}", e);
                        }
                    }
                    ActorInfo::GetState { response_tx } => {
                        match &*actor_instance_wrapper.read().await {
                            Some(instance) => {
                                let state = instance.actor_store.get_state();
                                if let Err(e) = response_tx.send(Ok(state)) {
                                    error!("Failed to send state response: {:?}", e);
                                }
                            }
                            None => {
                                let err =
                                    ActorError::UnexpectedError("Actor instance not found".to_string());
                                if let Err(e) = response_tx.send(Err(err)) {
                                    error!("Failed to send state error response: {:?}", e);
                                }
                            }
                        }
                    }
                    ActorInfo::GetChain { response_tx } => {
                        match &*actor_instance_wrapper.read().await {
                            None => {
                                let err =
                                    ActorError::UnexpectedError("Actor instance not found".to_string());
                                if let Err(e) = response_tx.send(Err(err)) {
                                    error!("Failed to send chain error response: {:?}", e);
                                }
                                return;
                            }
                            Some(instance) => {
                                let chain = instance.actor_store.get_chain();
                                if let Err(e) = response_tx.send(Ok(chain)) {
                                    error!("Failed to send chain response: {:?}", e);
                                }
                            }
                        };
                    }
                    ActorInfo::GetMetrics { response_tx } => {
                        let metrics = metrics.read().await;
                        let metrics_data = metrics.get_metrics().await;
                        if let Err(e) = response_tx.send(Ok(metrics_data)) {
                            error!("Failed to send metrics response: {:?}", e);
                        }
                    }
                    ActorInfo::SaveChain { response_tx } => {
                        match &*actor_instance_wrapper.read().await {
                            None => {
                                let err =
                                    ActorError::UnexpectedError("Actor instance not found".to_string());
                                if let Err(e) = response_tx.send(Err(err)) {
                                    error!("Failed to send save chain error response: {:?}", e);
                                }
                                return;
                            }
                            Some(instance) => match instance.actor_store.save_chain() {
                                Ok(_) => {
                                    if let Err(e) = response_tx.send(Ok(())) {
                                        error!("Failed to send save chain response: {:?}", e);
                                    }
                                }
                                Err(e) => {
                                    if let Err(send_err) = response_tx
                                        .send(Err(ActorError::UnexpectedError(e.to_string())))
                                    {
                                        error!(
                                            "Failed to send save chain error response: {:?}",
                                            send_err
                                        );
                                    }
                                }
                            },
                        };
                    }
                    ActorInfo::GetExportHashes { response_tx } => {
                        match &mut *actor_instance_wrapper.write().await {
                            Some(instance) => {
                                match instance.get_export_hashes().await {
                                    Ok(hashes) => {
                                        if let Err(e) = response_tx.send(Ok(hashes)) {
                                            error!("Failed to send export hashes response: {:?}", e);
                                        }
                                    }
                                    Err(e) => {
                                        let err = ActorError::UnexpectedError(format!(
                                            "Failed to get export hashes: {}",
                                            e
                                        ));
                                        if let Err(e) = response_tx.send(Err(err)) {
                                            error!("Failed to send export hashes error response: {:?}", e);
                                        }
                                    }
                                }
                            }
                            None => {
                                let err = ActorError::UnexpectedError("Actor instance not found".to_string());
                                if let Err(e) = response_tx.send(Err(err)) {
                                    error!("Failed to send export hashes error response: {:?}", e);
                                }
                            }
                        }
                    }
                }  // close match info
            }  // close Some(info) branch

                    else => break,
                } // close select!
        }
    }

    /// Execute a function call using Pack-native Value encoding.
    ///
    /// Decodes params from Pack ABI bytes to a structured `Value`,
    /// then calls `call_function_with_value` which preserves type information.
    async fn execute_call_pack(
        actor_instance: &mut PackInstance,
        name: &String,
        params: Vec<u8>,
        _theater_tx: &mpsc::Sender<TheaterCommand>,
        metrics: &MetricsCollector,
    ) -> Result<Vec<u8>, ActorError> {
        let start = Instant::now();

        let state = actor_instance.actor_store.get_state();
        debug!(
            "Executing pack call to function '{}' with state: {:?}",
            name, state
        );

        let params_value = crate::pack_bridge::decode_value(&params).map_err(|e| {
            error!("Failed to decode pack params: {}", e);
            ActorError::SerializationError
        })?;

        actor_instance.actor_store.record_event(ChainEventData {
            event_type: "wasm".to_string(),
            data: WasmEventData::WasmCall {
                function_name: name.clone(),
                params: params.clone(),
            }
            .into(),
        });

        let (new_state, results) = match actor_instance
            .call_function_with_value(name, state, params_value)
            .await
        {
            Ok(result) => {
                actor_instance.actor_store.record_event(ChainEventData {
                    event_type: "wasm".to_string(),
                    data: WasmEventData::WasmResult {
                        function_name: name.clone(),
                        result: result.clone(),
                    }
                    .into(),
                });
                result
            }
            Err(e) => {
                let event = actor_instance.actor_store.record_event(ChainEventData {
                    event_type: "wasm".to_string(),
                    data: WasmEventData::WasmError {
                        function_name: name.clone(),
                        message: e.to_string(),
                    }
                    .into(),
                });

                error!("Failed to execute function '{}': {}", name, e);
                return Err(ActorError::Internal(event));
            }
        };

        debug!("Pack call to '{}' completed", name);
        actor_instance.actor_store.set_state(new_state);

        let duration = start.elapsed();
        metrics.record_operation(duration, true).await;

        Ok(results)
    }
}
