blvm-node 0.1.88

//! IPC server for node
//!
//! Server-side IPC implementation that the node uses to communicate with modules.
//! Handles incoming connections from module processes.
//! Unix: Unix domain sockets. Windows: named pipes.

use futures::{SinkExt, StreamExt};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::sync::mpsc;
use tokio_util::codec::{FramedRead, FramedWrite, LengthDelimitedCodec};
use tracing::{debug, error, info, warn};

use async_trait::async_trait;

use crate::module::api::events::EventManager;
use crate::module::api::hub::ModuleApiHub;
use crate::module::ipc::module_ipc_length_codec;
use crate::module::ipc::protocol::{
    CliSpec, InvocationMessage, InvocationResultMessage, InvocationResultPayload, InvocationType,
    ModuleMessage, RequestMessage, RequestPayload, ResponseMessage, ResponsePayload,
};
use crate::module::traits::{EventType, ModuleError, NodeAPI, module_error_msg};
use tokio::sync::oneshot;

/// Async invoker for in-process WASM modules. Used when a module has no IPC connection.
#[cfg(feature = "wasm-modules")]
#[async_trait]
pub trait WasmInvoker: Send + Sync {
    async fn invoke_cli(
        &self,
        module_name: &str,
        subcommand: &str,
        args: Vec<String>,
    ) -> Result<InvocationResultPayload, ModuleError>;

    async fn invoke_rpc(
        &self,
        module_name: &str,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value, ModuleError>;

    async fn on_event(&self, module_name: &str, event_json: &str) -> Result<String, ModuleError>;
}

/// IPC server that handles module connections
pub struct ModuleIpcServer {
    /// Socket/pipe path where server listens
    socket_path: PathBuf,
    /// Connection count (for fallback module IDs)
    connection_count: Arc<std::sync::atomic::AtomicUsize>,
    /// Event manager for publishing events
    event_manager: Option<Arc<crate::module::api::events::EventManager>>,
    /// API hub for request routing
    api_hub: Option<Arc<tokio::sync::Mutex<ModuleApiHub>>>,
    /// RPC request channels (module_id -> channel) for RPC endpoint registration
    /// Channel: (correlation_id, method, params, response_tx)
    rpc_channels: Arc<
        tokio::sync::RwLock<
            HashMap<
                String,
                mpsc::UnboundedSender<(
                    u64,
                    String,
                    serde_json::Value,
                    mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
                )>,
            >,
        >,
    >,
    /// CLI specs (module_id -> spec) registered by modules on connect
    cli_registry: Arc<tokio::sync::RwLock<HashMap<String, CliSpec>>>,
    /// Outgoing channel per module (for invoke_cli; module_id -> tx)
    outgoing_tx_by_module:
        Arc<tokio::sync::RwLock<HashMap<String, mpsc::UnboundedSender<bytes::Bytes>>>>,
    /// Pending CLI/RPC invocations (correlation_id -> response sender)
    pending_invocations:
        Arc<tokio::sync::Mutex<HashMap<u64, oneshot::Sender<InvocationResultMessage>>>>,
    /// Pending RPC responses from IpcRpcHandler (correlation_id -> response sender)
    rpc_pending: Arc<
        tokio::sync::Mutex<
            HashMap<
                u64,
                mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
            >,
        >,
    >,
    /// Next correlation ID for invocations
    next_invocation_id: Arc<AtomicU64>,
    /// In-process WASM invoker (when module has no IPC connection)
    #[cfg(feature = "wasm-modules")]
    wasm_invoker: Option<Arc<dyn WasmInvoker>>,
    /// Timer service notified when a module connects and disconnects.
    timer_service: Option<crate::module::timers::TimerService>,
}

/// Active connection to a module (generic over stream read half)
struct ModuleConnection<R: tokio::io::AsyncRead> {
    /// Module ID
    module_id: String,
    /// Framed reader for receiving messages
    reader: FramedRead<R, LengthDelimitedCodec>,
    /// Channel for sending outgoing messages (responses and events)
    outgoing_tx: Option<mpsc::UnboundedSender<bytes::Bytes>>,
    /// Event subscriptions for this module
    subscriptions: Vec<EventType>,
    /// Event channel sender for this module (used by EventManager)
    event_tx: Option<mpsc::Sender<ModuleMessage>>,
    /// Handle to the unified writer task
    writer_task_handle: Option<tokio::task::JoinHandle<()>>,
    /// Channel for RPC requests to this module (for module RPC endpoints)
    /// (correlation_id, method, params, response_tx)
    rpc_request_tx: Option<
        mpsc::UnboundedSender<(
            u64,
            String,
            serde_json::Value,
            mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
        )>,
    >,
}

impl<R: tokio::io::AsyncRead> Drop for ModuleConnection<R> {
    fn drop(&mut self) {
        // The read loop aborts this on a normal disconnect. Cancellation of the
        // connection task skips that path, and dropping the handle leaves the
        // write half open, so the client blocks instead of seeing a closed socket.
        if let Some(handle) = self.writer_task_handle.take() {
            handle.abort();
        }
    }
}

/// Abort accepted-connection tasks when the accept loop is dropped.
struct AbortOnDrop(Vec<tokio::task::JoinHandle<()>>);

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        for handle in &self.0 {
            handle.abort();
        }
    }
}

/// Lock-free handle to IPC server state needed for cross-task invocations.
///
/// Holding the full `Arc<Mutex<ModuleIpcServer>>` across an await would deadlock
/// because `start()` holds that Mutex for its entire lifetime.  This struct carries
/// only the individually `Arc`-wrapped fields required by `invoke_module_api`, so
/// callers (e.g. `IpcForwardingModuleAPI`) can operate concurrently without
/// contending with the server loop.
#[derive(Clone)]
pub struct ModuleIpcHandle {
    pub(crate) outgoing_tx_by_module:
        Arc<tokio::sync::RwLock<HashMap<String, mpsc::UnboundedSender<bytes::Bytes>>>>,
    pub(crate) pending_invocations:
        Arc<tokio::sync::Mutex<HashMap<u64, oneshot::Sender<InvocationResultMessage>>>>,
    pub(crate) next_invocation_id: Arc<AtomicU64>,
    pub(crate) cli_registry: Arc<tokio::sync::RwLock<HashMap<String, CliSpec>>>,
}

impl ModuleIpcHandle {
    /// Register a CLI spec without locking the IPC server (the accept loop holds that mutex).
    pub async fn register_cli_spec(&self, module_id: String, spec: CliSpec) {
        self.cli_registry.write().await.insert(module_id, spec);
    }

    /// Drop a CLI spec registered under `module_name`.
    pub async fn unregister_cli_spec_by_name(&self, module_name: &str) {
        let mut registry = self.cli_registry.write().await;
        registry.retain(|id, _| id != module_name && !id.starts_with(&format!("{module_name}_")));
    }

    /// Invoke a subprocess module's registered ModuleAPI method (node → module).
    pub async fn invoke_module_api(
        &self,
        module_id: &str,
        method: &str,
        params: &[u8],
        caller_module_id: &str,
    ) -> Result<Vec<u8>, ModuleError> {
        let outgoing_tx = {
            let by_module = self.outgoing_tx_by_module.read().await;
            by_module.get(module_id).cloned()
        };
        let Some(outgoing_tx) = outgoing_tx else {
            return Err(ModuleError::OperationError(format!(
                "Module '{module_id}' is not connected"
            )));
        };

        let correlation_id = self.next_invocation_id.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        {
            let mut pending = self.pending_invocations.lock().await;
            pending.insert(correlation_id, tx);
        }

        let invocation = InvocationMessage {
            correlation_id,
            invocation_type: InvocationType::ModuleApi {
                method: method.to_string(),
                params: params.to_vec(),
                caller_module_id: caller_module_id.to_string(),
            },
        };
        let bytes = bincode::serialize(&ModuleMessage::Invocation(invocation))
            .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

        if outgoing_tx.send(bytes::Bytes::from(bytes)).is_err() {
            self.pending_invocations
                .lock()
                .await
                .remove(&correlation_id);
            return Err(ModuleError::OperationError(
                module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
            ));
        }

        let result = match tokio::time::timeout(tokio::time::Duration::from_secs(60), rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::OperationError(
                    module_error_msg::INVOCATION_RESPONSE_CHANNEL_CLOSED.to_string(),
                ));
            }
            Err(_) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::OperationError(
                    "Module did not respond to ModuleAPI invocation within 60 seconds".to_string(),
                ));
            }
        };

        if !result.success {
            return Err(ModuleError::OperationError(
                result.error.unwrap_or_else(|| "Unknown error".to_string()),
            ));
        }

        match result.payload {
            Some(InvocationResultPayload::ModuleApi(data)) => Ok(data),
            Some(_) => Err(ModuleError::OperationError(
                "Module returned wrong payload type for ModuleAPI invocation".to_string(),
            )),
            None => Err(ModuleError::OperationError(
                module_error_msg::MODULE_RETURNED_SUCCESS_BUT_NO_PAYLOAD.to_string(),
            )),
        }
    }

    /// Deliver a timer fire and wait for `InvocationResultPayload::Timer`.
    ///
    /// The pending entry is removed on send failure and on timeout, so a late
    /// result finds nothing to wake and is ignored.
    pub async fn invoke_timer(
        &self,
        module_id: &str,
        fire: crate::module::timers::types::TimerFire,
        ack_timeout: std::time::Duration,
    ) -> Result<crate::module::timers::types::TimerAck, ModuleError> {
        let outgoing_tx = {
            let by_module = self.outgoing_tx_by_module.read().await;
            by_module.get(module_id).cloned()
        };
        let Some(outgoing_tx) = outgoing_tx else {
            return Err(ModuleError::OperationError(
                module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
            ));
        };

        let correlation_id = self.next_invocation_id.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.pending_invocations
            .lock()
            .await
            .insert(correlation_id, tx);

        let invocation = InvocationMessage {
            correlation_id,
            invocation_type: InvocationType::Timer(fire),
        };
        let bytes = bincode::serialize(&ModuleMessage::Invocation(invocation))
            .map_err(|e| ModuleError::SerializationError(e.to_string()))?;
        if outgoing_tx.send(bytes::Bytes::from(bytes)).is_err() {
            self.pending_invocations
                .lock()
                .await
                .remove(&correlation_id);
            return Err(ModuleError::OperationError(
                module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
            ));
        }

        let result = match tokio::time::timeout(ack_timeout, rx).await {
            Ok(Ok(result)) => result,
            Ok(Err(_)) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::OperationError(
                    module_error_msg::INVOCATION_RESPONSE_CHANNEL_CLOSED.to_string(),
                ));
            }
            Err(_) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::Timeout);
            }
        };
        if !result.success {
            return Err(ModuleError::OperationError(
                result
                    .error
                    .unwrap_or_else(|| "timer fire failed".to_string()),
            ));
        }
        match result.payload {
            Some(InvocationResultPayload::Timer(ack)) => Ok(ack),
            _ => Err(ModuleError::OperationError(
                "module returned no timer acknowledgement".to_string(),
            )),
        }
    }

    /// Send one invocation and wait. Callers that own one connection send the
    /// next frame only after this returns, so that connection stays ordered.
    pub async fn invoke_hosted(
        &self,
        module_id: &str,
        invocation_type: InvocationType,
        timeout: std::time::Duration,
    ) -> Result<InvocationResultMessage, ModuleError> {
        let outgoing_tx = {
            let by_module = self.outgoing_tx_by_module.read().await;
            by_module.get(module_id).cloned()
        };
        let Some(outgoing_tx) = outgoing_tx else {
            return Err(ModuleError::OperationError(
                module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
            ));
        };

        let correlation_id = self.next_invocation_id.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = oneshot::channel();
        self.pending_invocations
            .lock()
            .await
            .insert(correlation_id, tx);

        let invocation = InvocationMessage {
            correlation_id,
            invocation_type,
        };
        let bytes = bincode::serialize(&ModuleMessage::Invocation(invocation))
            .map_err(|e| ModuleError::SerializationError(e.to_string()))?;
        if outgoing_tx.send(bytes::Bytes::from(bytes)).is_err() {
            self.pending_invocations
                .lock()
                .await
                .remove(&correlation_id);
            return Err(ModuleError::OperationError(
                module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
            ));
        }

        match tokio::time::timeout(timeout, rx).await {
            Ok(Ok(result)) => Ok(result),
            Ok(Err(_)) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                Err(ModuleError::OperationError(
                    module_error_msg::INVOCATION_RESPONSE_CHANNEL_CLOSED.to_string(),
                ))
            }
            Err(_) => {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                Err(ModuleError::Timeout)
            }
        }
    }

    #[cfg(test)]
    pub(crate) async fn pending_invocation_count(&self) -> usize {
        self.pending_invocations.lock().await.len()
    }
}

impl ModuleIpcServer {
    /// Wake a waiting invoke, or ignore the result when the correlation is already gone.
    pub(crate) async fn complete_pending_invocation(&self, result: InvocationResultMessage) {
        let mut pending = self.pending_invocations.lock().await;
        if let Some(tx) = pending.remove(&result.correlation_id) {
            let _ = tx.send(result);
        }
    }

    /// Return a lock-free handle for cross-task invocations (avoids deadlocking on the server Mutex).
    pub fn handle(&self) -> ModuleIpcHandle {
        ModuleIpcHandle {
            outgoing_tx_by_module: Arc::clone(&self.outgoing_tx_by_module),
            pending_invocations: Arc::clone(&self.pending_invocations),
            next_invocation_id: Arc::clone(&self.next_invocation_id),
            cli_registry: Arc::clone(&self.cli_registry),
        }
    }

    /// Create a new IPC server
    pub fn new<P: AsRef<Path>>(socket_path: P) -> Self {
        Self {
            socket_path: socket_path.as_ref().to_path_buf(),
            connection_count: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            event_manager: None,
            api_hub: None,
            rpc_channels: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            cli_registry: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            outgoing_tx_by_module: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            pending_invocations: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            rpc_pending: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            next_invocation_id: Arc::new(AtomicU64::new(1)),
            #[cfg(feature = "wasm-modules")]
            wasm_invoker: None,
            timer_service: None,
        }
    }

    /// Notify this service when modules connect and disconnect.
    pub fn with_timer_service(
        mut self,
        timer_service: crate::module::timers::TimerService,
    ) -> Self {
        self.timer_service = Some(timer_service);
        self
    }

    /// Set the WASM invoker for in-process module dispatch
    #[cfg(feature = "wasm-modules")]
    pub fn with_wasm_invoker(mut self, invoker: Arc<dyn WasmInvoker>) -> Self {
        self.wasm_invoker = Some(invoker);
        self
    }

    /// Get RPC request channel for a module (for RPC endpoint registration)
    pub async fn get_rpc_channel(
        &self,
        module_id: &str,
    ) -> Option<
        mpsc::UnboundedSender<(
            u64,
            String,
            serde_json::Value,
            mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
        )>,
    > {
        let channels = self.rpc_channels.read().await;
        channels.get(module_id).cloned()
    }

    /// Get all registered CLI specs (module_id -> spec)
    pub async fn get_cli_specs(&self) -> HashMap<String, CliSpec> {
        let registry = self.cli_registry.read().await;
        registry.clone()
    }

    /// Register CLI spec for a module (e.g. WASM modules that don't connect via IPC).
    pub async fn register_cli_spec(&self, module_id: String, spec: CliSpec) {
        let mut registry = self.cli_registry.write().await;
        registry.insert(module_id, spec);
    }

    /// Drop the CLI spec stored under `module_id`.
    pub async fn unregister_cli_spec(&self, module_id: &str) {
        let mut registry = self.cli_registry.write().await;
        registry.remove(module_id);
    }

    /// Unregister CLI spec by module name (for WASM and native; removes any entry with matching spec.name).
    pub async fn unregister_cli_spec_by_name(&self, module_name: &str) {
        let mut registry = self.cli_registry.write().await;
        let to_remove: Vec<String> = registry
            .iter()
            .filter(|(_, spec)| spec.name == module_name)
            .map(|(id, _)| id.clone())
            .collect();
        for id in to_remove {
            registry.remove(&id);
        }
    }

    /// Invoke a module CLI subcommand (node → module)
    ///
    /// Returns the invocation result or an error if the module is not found or does not respond.
    /// Tries IPC first for native modules; falls back to in-process WASM if configured.
    pub async fn invoke_cli(
        &self,
        module_name: &str,
        subcommand: &str,
        args: Vec<String>,
    ) -> Result<InvocationResultPayload, ModuleError> {
        // Try IPC path: find module_id by CLI spec name and get outgoing channel
        let maybe_ipc = {
            let registry = self.cli_registry.read().await;
            let module_id = registry
                .iter()
                .find(|(_, spec)| spec.name == module_name)
                .map(|(id, _)| id.clone());
            let by_module = self.outgoing_tx_by_module.read().await;
            module_id.and_then(|id| by_module.get(&id).cloned())
        };

        if let Some(outgoing_tx) = maybe_ipc {
            let correlation_id = self.next_invocation_id.fetch_add(1, Ordering::SeqCst);
            let (tx, rx) = oneshot::channel();

            {
                let mut pending = self.pending_invocations.lock().await;
                pending.insert(correlation_id, tx);
            }

            let invocation = InvocationMessage {
                correlation_id,
                invocation_type: InvocationType::Cli {
                    subcommand: subcommand.to_string(),
                    args,
                },
            };
            let bytes = bincode::serialize(&ModuleMessage::Invocation(invocation))
                .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

            if outgoing_tx.send(bytes::Bytes::from(bytes)).is_err() {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::OperationError(
                    module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
                ));
            }

            let result = match tokio::time::timeout(tokio::time::Duration::from_secs(60), rx).await
            {
                Ok(Ok(result)) => result,
                Ok(Err(_)) => {
                    self.pending_invocations
                        .lock()
                        .await
                        .remove(&correlation_id);
                    return Err(ModuleError::OperationError(
                        module_error_msg::INVOCATION_RESPONSE_CHANNEL_CLOSED.to_string(),
                    ));
                }
                Err(_) => {
                    self.pending_invocations
                        .lock()
                        .await
                        .remove(&correlation_id);
                    return Err(ModuleError::OperationError(
                        module_error_msg::MODULE_DID_NOT_RESPOND_CLI_60S.to_string(),
                    ));
                }
            };

            if !result.success {
                return Err(ModuleError::Cli(
                    result.error.unwrap_or_else(|| "Unknown error".to_string()),
                ));
            }

            return result.payload.ok_or_else(|| {
                ModuleError::OperationError(
                    module_error_msg::MODULE_RETURNED_SUCCESS_BUT_NO_PAYLOAD.to_string(),
                )
            });
        }

        // Fallback: in-process WASM invocation
        #[cfg(feature = "wasm-modules")]
        if let Some(ref invoker) = self.wasm_invoker {
            return invoker.invoke_cli(module_name, subcommand, args).await;
        }

        Err(ModuleError::OperationError(format!(
            "No module with CLI name '{module_name}' is loaded"
        )))
    }

    /// Invoke a module RPC method (node → module).
    ///
    /// Finds module by CLI spec name (same as invoke_cli). Returns the JSON result or an error.
    /// Tries IPC first for native modules; falls back to in-process WASM if configured.
    pub async fn invoke_rpc(
        &self,
        module_name: &str,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value, ModuleError> {
        // Try IPC path
        let maybe_ipc = {
            let registry = self.cli_registry.read().await;
            let module_id = registry
                .iter()
                .find(|(_, spec)| spec.name == module_name)
                .map(|(id, _)| id.clone());
            let by_module = self.outgoing_tx_by_module.read().await;
            module_id.and_then(|id| by_module.get(&id).cloned())
        };

        if let Some(outgoing_tx) = maybe_ipc {
            let correlation_id = self.next_invocation_id.fetch_add(1, Ordering::SeqCst);
            let (tx, rx) = oneshot::channel();

            {
                let mut pending = self.pending_invocations.lock().await;
                pending.insert(correlation_id, tx);
            }

            let invocation = InvocationMessage {
                correlation_id,
                invocation_type: InvocationType::Rpc {
                    method: method.to_string(),
                    params,
                },
            };
            let bytes = bincode::serialize(&ModuleMessage::Invocation(invocation))
                .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

            if outgoing_tx.send(bytes::Bytes::from(bytes)).is_err() {
                self.pending_invocations
                    .lock()
                    .await
                    .remove(&correlation_id);
                return Err(ModuleError::OperationError(
                    module_error_msg::MODULE_CONNECTION_CLOSED.to_string(),
                ));
            }

            let result = match tokio::time::timeout(tokio::time::Duration::from_secs(60), rx).await
            {
                Ok(Ok(result)) => result,
                Ok(Err(_)) => {
                    self.pending_invocations
                        .lock()
                        .await
                        .remove(&correlation_id);
                    return Err(ModuleError::OperationError(
                        module_error_msg::INVOCATION_RESPONSE_CHANNEL_CLOSED.to_string(),
                    ));
                }
                Err(_) => {
                    self.pending_invocations
                        .lock()
                        .await
                        .remove(&correlation_id);
                    return Err(ModuleError::OperationError(
                        module_error_msg::MODULE_DID_NOT_RESPOND_RPC_60S.to_string(),
                    ));
                }
            };

            if !result.success {
                return Err(ModuleError::OperationError(
                    result.error.unwrap_or_else(|| "Unknown error".to_string()),
                ));
            }

            match result.payload {
                Some(InvocationResultPayload::Rpc(value)) => return Ok(value),
                Some(_) => {
                    return Err(ModuleError::OperationError(
                        module_error_msg::MODULE_RETURNED_WRONG_PAYLOAD_TYPE_RPC.to_string(),
                    ));
                }
                None => {
                    return Err(ModuleError::OperationError(
                        module_error_msg::MODULE_RETURNED_SUCCESS_BUT_NO_PAYLOAD.to_string(),
                    ));
                }
            }
        }

        // Fallback: in-process WASM invocation
        #[cfg(feature = "wasm-modules")]
        if let Some(ref invoker) = self.wasm_invoker {
            return invoker.invoke_rpc(module_name, method, params).await;
        }

        Err(ModuleError::OperationError(format!(
            "No module with CLI name '{module_name}' is loaded"
        )))
    }

    /// Set event manager for publishing events
    pub fn with_event_manager(mut self, event_manager: Arc<EventManager>) -> Self {
        self.event_manager = Some(event_manager);
        self
    }

    /// Set API hub for request routing
    pub fn with_api_hub(mut self, api_hub: Arc<tokio::sync::Mutex<ModuleApiHub>>) -> Self {
        self.api_hub = Some(api_hub);
        self
    }

    /// Start listening for module connections
    pub async fn start<A: NodeAPI + Send + Sync + 'static>(
        &mut self,
        node_api: Arc<A>,
    ) -> Result<(), ModuleError> {
        #[cfg(unix)]
        return self.start_unix(node_api).await;
        #[cfg(windows)]
        return self.start_windows(node_api).await;
    }

    #[cfg(unix)]
    async fn start_unix<A: NodeAPI + Send + Sync + 'static>(
        &mut self,
        node_api: Arc<A>,
    ) -> Result<(), ModuleError> {
        use tokio::net::{UnixListener, UnixStream};

        if self.socket_path.exists() {
            std::fs::remove_file(&self.socket_path)
                .map_err(|e| ModuleError::IpcError(format!("Failed to remove old socket: {e}")))?;
        }
        if let Some(parent) = self.socket_path.parent() {
            std::fs::create_dir_all(parent).map_err(|e| {
                ModuleError::IpcError(format!("Failed to create socket directory: {e}"))
            })?;
        }
        let listener = UnixListener::bind(&self.socket_path)
            .map_err(|e| ModuleError::IpcError(format!("Failed to bind socket: {e}")))?;
        {
            use std::os::unix::fs::PermissionsExt;
            let perms = std::fs::Permissions::from_mode(0o600);
            if let Err(e) = std::fs::set_permissions(&self.socket_path, perms) {
                warn!("Failed to set IPC socket permissions: {}", e);
            }
        }
        info!("Module IPC server listening on {:?}", self.socket_path);
        let ctx = self.connection_ctx();
        let mut connections = AbortOnDrop(Vec::new());
        loop {
            match listener.accept().await {
                Ok((stream, _)) => {
                    debug!("New module connection");
                    let ctx = ctx.clone();
                    let node_api = Arc::clone(&node_api);
                    connections.0.push(tokio::spawn(async move {
                        if let Err(e) = ctx.handle_connection(stream, node_api).await {
                            error!("Module connection closed: {e}");
                        }
                    }));
                }
                Err(e) => error!("Failed to accept module connection: {}", e),
            }
        }
    }

    #[cfg(windows)]
    async fn start_windows<A: NodeAPI + Send + Sync + 'static>(
        &mut self,
        node_api: Arc<A>,
    ) -> Result<(), ModuleError> {
        use tokio::net::windows::named_pipe::ServerOptions;

        let pipe_name = path_to_pipe_name(&self.socket_path);
        if let Some(parent) = self.socket_path.parent() {
            std::fs::create_dir_all(parent).map_err(|e| {
                ModuleError::IpcError(format!("Failed to create socket directory: {e}"))
            })?;
        }
        info!("Module IPC server listening on pipe {}", pipe_name);
        let ctx = self.connection_ctx();
        let mut connections = AbortOnDrop(Vec::new());
        loop {
            let server = ServerOptions::new()
                .first_pipe_instance(false)
                .create(&pipe_name)
                .map_err(|e| ModuleError::IpcError(format!("Failed to create named pipe: {e}")))?;
            server
                .connect()
                .await
                .map_err(|e| ModuleError::IpcError(format!("Failed to connect pipe: {e}")))?;
            debug!("New module connection (named pipe)");
            let ctx = ctx.clone();
            let node_api = Arc::clone(&node_api);
            connections.0.push(tokio::spawn(async move {
                if let Err(e) = ctx.handle_connection(server, node_api).await {
                    error!("Module connection closed: {e}");
                }
            }));
        }
    }
}

/// Shared per-connection state. Each accepted module runs on its own task.
#[derive(Clone)]
struct ConnectionCtx {
    connection_count: Arc<std::sync::atomic::AtomicUsize>,
    event_manager: Option<Arc<crate::module::api::events::EventManager>>,
    api_hub: Option<Arc<tokio::sync::Mutex<ModuleApiHub>>>,
    rpc_channels: Arc<
        tokio::sync::RwLock<
            HashMap<
                String,
                mpsc::UnboundedSender<(
                    u64,
                    String,
                    serde_json::Value,
                    mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
                )>,
            >,
        >,
    >,
    cli_registry: Arc<tokio::sync::RwLock<HashMap<String, CliSpec>>>,
    outgoing_tx_by_module:
        Arc<tokio::sync::RwLock<HashMap<String, mpsc::UnboundedSender<bytes::Bytes>>>>,
    pending_invocations:
        Arc<tokio::sync::Mutex<HashMap<u64, oneshot::Sender<InvocationResultMessage>>>>,
    rpc_pending: Arc<
        tokio::sync::Mutex<
            HashMap<
                u64,
                mpsc::UnboundedSender<Result<serde_json::Value, crate::rpc::errors::RpcError>>,
            >,
        >,
    >,
    timer_service: Option<crate::module::timers::TimerService>,
}

impl ModuleIpcServer {
    fn connection_ctx(&self) -> ConnectionCtx {
        ConnectionCtx {
            connection_count: Arc::clone(&self.connection_count),
            event_manager: self.event_manager.clone(),
            api_hub: self.api_hub.clone(),
            rpc_channels: Arc::clone(&self.rpc_channels),
            cli_registry: Arc::clone(&self.cli_registry),
            outgoing_tx_by_module: Arc::clone(&self.outgoing_tx_by_module),
            pending_invocations: Arc::clone(&self.pending_invocations),
            rpc_pending: Arc::clone(&self.rpc_pending),
            timer_service: self.timer_service.clone(),
        }
    }
}

impl ConnectionCtx {
    async fn complete_pending_invocation(&self, result: InvocationResultMessage) {
        let mut pending = self.pending_invocations.lock().await;
        if let Some(tx) = pending.remove(&result.correlation_id) {
            let _ = tx.send(result);
        }
    }

    /// Handle a new module connection (generic over stream type)
    async fn handle_connection<S, A>(&self, stream: S, node_api: Arc<A>) -> Result<(), ModuleError>
    where
        S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
        A: NodeAPI + Send + Sync,
    {
        let (read_half, write_half) = tokio::io::split(stream);
        let mut reader = FramedRead::new(read_half, module_ipc_length_codec());
        let mut writer = FramedWrite::new(write_half, module_ipc_length_codec());

        // Wait for handshake message from module
        let (module_id, module_name) = match reader.next().await {
            Some(Ok(bytes)) => {
                let message: ModuleMessage = bincode::deserialize(bytes.as_ref())
                    .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

                match message {
                    ModuleMessage::Request(request) => {
                        let accepted = match &request.payload {
                            RequestPayload::Handshake {
                                module_id,
                                module_name,
                                version,
                            } => {
                                info!(
                                    "Module handshake: id={}, name={}, version={}",
                                    module_id, module_name, version
                                );
                                Some((module_id.clone(), module_name.clone()))
                            }
                            RequestPayload::HandshakeV2 {
                                module_id,
                                module_name,
                                version,
                                ipc_protocol,
                            } => {
                                if *ipc_protocol
                                    != crate::module::timers::types::IPC_PROTOCOL_VERSION
                                {
                                    return Err(ModuleError::IpcError(format!(
                                        "unsupported ipc protocol version {ipc_protocol}"
                                    )));
                                }
                                info!(
                                    "Module handshake v2: id={}, name={}, version={}, ipc={}",
                                    module_id, module_name, version, ipc_protocol
                                );
                                Some((module_id.clone(), module_name.clone()))
                            }
                            _ => None,
                        };
                        if let Some((module_id, module_name)) = accepted {
                            let ack = ResponseMessage {
                                correlation_id: request.correlation_id,
                                success: true,
                                payload: Some(ResponsePayload::HandshakeAck {
                                    node_version: env!("CARGO_PKG_VERSION").to_string(),
                                }),
                                error: None,
                            };
                            let ack_bytes = bincode::serialize(&ModuleMessage::Response(ack))
                                .map_err(|e| ModuleError::SerializationError(e.to_string()))?;
                            writer
                                .send(bytes::Bytes::from(ack_bytes))
                                .await
                                .map_err(|e| {
                                    ModuleError::IpcError(format!(
                                        "Failed to send handshake ack: {e}"
                                    ))
                                })?;
                            (module_id, module_name)
                        } else {
                            warn!("Module did not send handshake, using fallback ID");
                            let timestamp = crate::utils::current_timestamp_nanos();
                            let count = self.connection_count.fetch_add(1, Ordering::SeqCst);
                            let module_id = format!("module_{count}_{timestamp}");
                            (module_id.clone(), module_id)
                        }
                    }
                    _ => {
                        return Err(ModuleError::IpcError(
                            "First message must be a handshake request".to_string(),
                        ));
                    }
                }
            }
            Some(Err(e)) => {
                return Err(ModuleError::IpcError(format!(
                    "Failed to read handshake: {e}"
                )));
            }
            None => {
                return Err(ModuleError::IpcError(
                    "Connection closed before handshake".to_string(),
                ));
            }
        };

        // Create unified outgoing message channel (for both responses and events)
        // This allows us to share the writer between response handler and event handler
        let (outgoing_tx, mut outgoing_rx) = mpsc::unbounded_channel::<bytes::Bytes>();

        // Create RPC request channel (for sending RPC requests from node to module)
        // (correlation_id, method, params, response_tx)
        type RpcResponseSender = mpsc::UnboundedSender<
            std::result::Result<serde_json::Value, crate::rpc::errors::RpcError>,
        >;
        let (rpc_request_tx, mut rpc_request_rx) =
            mpsc::unbounded_channel::<(u64, String, serde_json::Value, RpcResponseSender)>();

        // Create event channel for this module (events from EventManager go here)
        let (event_tx, mut event_rx) = mpsc::channel(100);

        // Clone outgoing_tx before moving it into the task
        let outgoing_tx_for_events = outgoing_tx.clone();

        // Spawn unified writer task that handles both responses and events
        let module_id_writer_task = module_id.clone();
        let event_manager_clone = self.event_manager.clone();
        let writer_task_handle = tokio::spawn(async move {
            // Forward events from event_rx to outgoing_tx
            let module_id_event_fwd = module_id_writer_task.clone();
            tokio::spawn(async move {
                while let Some(event_message) = event_rx.recv().await {
                    match bincode::serialize(&event_message) {
                        Ok(bytes) => {
                            if outgoing_tx_for_events
                                .send(bytes::Bytes::from(bytes))
                                .is_err()
                            {
                                break; // Receiver dropped, connection closed
                            }
                        }
                        Err(e) => {
                            warn!(
                                "Failed to serialize event for module {}: {}",
                                module_id_event_fwd, e
                            );
                        }
                    }
                }

                // Clean up: unsubscribe module from events when task exits
                if let Some(event_mgr) = event_manager_clone {
                    if let Err(e) = event_mgr.unsubscribe_module(&module_id_event_fwd).await {
                        warn!(
                            "Failed to unsubscribe module {} from events: {}",
                            module_id_event_fwd, e
                        );
                    }
                }
            });

            // Main writer loop: send all outgoing messages (responses + events) via IPC
            while let Some(bytes) = outgoing_rx.recv().await {
                if let Err(e) = writer.send(bytes).await {
                    warn!(
                        "Failed to send message to module {}: {}",
                        module_id_writer_task, e
                    );
                    break;
                }
            }
        });

        // Module name comes from the handshake. Fallback ids have no 32-hex suffix.

        // Get base data directory - extract from module_id or use default
        // Module ID format: {module_name}_{uuid}
        // We'll derive the base directory from the module name
        // Default module data root when ModuleManager does not supply one.
        let base_data_dir = std::path::PathBuf::from("data/modules");
        let module_data_dir = base_data_dir.join(&module_name);

        // Ensure module data directory exists
        if let Err(e) = std::fs::create_dir_all(&module_data_dir) {
            warn!(
                "Failed to create module data directory {:?}: {}",
                module_data_dir, e
            );
        }

        // Initialize module filesystem and storage access
        if let Err(e) = node_api
            .initialize_module(module_id.clone(), module_data_dir, base_data_dir)
            .await
        {
            warn!(
                "Failed to initialize module {} filesystem/storage: {}",
                module_id, e
            );
            // Continue anyway - module can still use other APIs
        }

        // Register for invoke_cli and RPC dispatch (before moving outgoing_tx)
        {
            let mut by_module = self.outgoing_tx_by_module.write().await;
            by_module.insert(module_id.clone(), outgoing_tx.clone());
        }
        {
            let mut channels = self.rpc_channels.write().await;
            channels.insert(module_id.clone(), rpc_request_tx.clone());
        }
        if let Some(service) = &self.timer_service {
            service.attach_module(&module_name, &module_id).await;
        }

        // Spawn task to forward RPC requests from IpcRpcHandler to module via Invocation
        let outgoing_tx_for_rpc = outgoing_tx.clone();
        let rpc_pending = Arc::clone(&self.rpc_pending);
        tokio::spawn(async move {
            while let Some((correlation_id, method, params, response_tx)) =
                rpc_request_rx.recv().await
            {
                rpc_pending.lock().await.insert(correlation_id, response_tx);

                let invocation = InvocationMessage {
                    correlation_id,
                    invocation_type: InvocationType::Rpc { method, params },
                };
                if let Ok(bytes) = bincode::serialize(&ModuleMessage::Invocation(invocation)) {
                    if outgoing_tx_for_rpc.send(bytes::Bytes::from(bytes)).is_err() {
                        break; // Connection closed
                    }
                }
            }
        });

        let mut connection = ModuleConnection {
            module_id: module_id.clone(),
            reader,
            outgoing_tx: Some(outgoing_tx),
            subscriptions: Vec::new(),
            event_tx: Some(event_tx),
            writer_task_handle: Some(writer_task_handle),
            rpc_request_tx: Some(rpc_request_tx),
        };

        // Process messages from this module
        while let Some(result) = connection.reader.next().await {
            match result {
                Ok(bytes) => {
                    let node_api_clone = Arc::clone(&node_api);
                    match self
                        .handle_message(bytes.as_ref(), &mut connection, node_api_clone)
                        .await
                    {
                        Ok(()) => {}
                        Err(e) => {
                            error!("Error handling message: {}", e);
                            break;
                        }
                    }
                }
                Err(e) => {
                    error!("Error reading from module {}: {}", module_id, e);
                    break;
                }
            }
        }

        info!("Module {} disconnected", module_id);

        if let Some(service) = &self.timer_service {
            service.detach_module(&module_name, Some(&module_id)).await;
        }

        let _ = node_api.unregister_subprocess_module_api(&module_id).await;

        // Clean up: remove from shared maps
        {
            let mut by_module = self.outgoing_tx_by_module.write().await;
            by_module.remove(&module_id);
        }
        {
            let mut channels = self.rpc_channels.write().await;
            channels.remove(&module_id);
        }
        {
            let mut registry = self.cli_registry.write().await;
            registry.remove(&module_id);
        }

        // Close outgoing channel (will cause writer task to exit)
        drop(connection.outgoing_tx.take());

        // Abort writer task (which includes event forwarding)
        if let Some(handle) = connection.writer_task_handle.take() {
            handle.abort();
        }

        // Unsubscribe from event manager
        if let Some(event_mgr) = &self.event_manager {
            if let Err(e) = event_mgr.unsubscribe_module(&module_id).await {
                warn!(
                    "Failed to unsubscribe module {} from events: {}",
                    module_id, e
                );
            }
        }

        Ok(())
    }

    /// Handle a message from a module
    async fn handle_message<R: tokio::io::AsyncRead, A: NodeAPI + Send + Sync>(
        &self,
        bytes: &[u8],
        connection: &mut ModuleConnection<R>,
        node_api: Arc<A>,
    ) -> Result<(), ModuleError> {
        let message: ModuleMessage = bincode::deserialize(bytes)
            .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

        match message {
            ModuleMessage::Request(request) => {
                // Handle RegisterCliSpec: store in registry, respond, don't pass to hub
                if let RequestPayload::RegisterCliSpec { spec } = &request.payload {
                    let module_id = connection.module_id.clone();
                    let mut registry = self.cli_registry.write().await;
                    registry.insert(module_id.clone(), spec.clone());
                    debug!("Module {} registered CLI spec: {}", module_id, spec.name);
                    let response = ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::Bool(true),
                    );
                    let response_message = ModuleMessage::Response(response);
                    let response_bytes = bincode::serialize(&response_message)
                        .map_err(|e| ModuleError::SerializationError(e.to_string()))?;
                    if let Some(tx) = &connection.outgoing_tx {
                        tx.send(bytes::Bytes::from(response_bytes)).map_err(|e| {
                            ModuleError::IpcError(format!("Failed to send response: {e}"))
                        })?;
                    }
                    return Ok(());
                }

                // Handle SubscribeEvents specially to register with event manager
                if let RequestPayload::SubscribeModuleEventNames { ref names } = request.payload {
                    if let Some(event_mgr) = &self.event_manager {
                        event_mgr
                            .set_module_event_filter(&connection.module_id, names.clone())
                            .await;
                    }
                }

                if let RequestPayload::SubscribeEvents { ref event_types } = request.payload {
                    if let Some(event_mgr) = &self.event_manager {
                        if let Some(event_tx) = &connection.event_tx {
                            // Register module subscriptions
                            let module_id = connection.module_id.clone();
                            let event_tx_clone = event_tx.clone();
                            event_mgr
                                .subscribe_module(
                                    module_id.clone(),
                                    event_types.clone(),
                                    event_tx_clone,
                                )
                                .await?;
                            connection.subscriptions = event_types.clone();
                            debug!(
                                "Module {} subscribed to events: {:?}",
                                module_id, event_types
                            );
                            // Note: subscribe_module() handles:
                            // 1. Sending already-loaded modules to this newly subscribing module
                            // 2. Publishing ModuleLoaded for this module (if loaded) AFTER subscription
                            // This ensures ModuleLoaded only happens after startup is complete
                        }
                    }
                }

                // Use API hub if available, otherwise fall back to direct node_api
                let response = if let Some(hub) = &self.api_hub {
                    let mut hub_guard = hub.lock().await;
                    hub_guard
                        .handle_request(&connection.module_id, request.clone())
                        .await?
                } else {
                    self.process_request(&request, node_api, &connection.module_id)
                        .await?
                };
                let response_message = ModuleMessage::Response(response);

                let response_bytes = bincode::serialize(&response_message)
                    .map_err(|e| ModuleError::SerializationError(e.to_string()))?;

                // Send response through outgoing channel
                if let Some(tx) = &connection.outgoing_tx {
                    tx.send(bytes::Bytes::from(response_bytes)).map_err(|e| {
                        ModuleError::IpcError(format!("Failed to send response: {e}"))
                    })?;
                }
            }
            ModuleMessage::Response(_) => {
                warn!("Received response from module (unexpected)");
            }
            ModuleMessage::Event(_) => {
                warn!("Received event from module (unexpected)");
            }
            ModuleMessage::InvocationResult(result) => {
                // Check rpc_pending first (IpcRpcHandler path)
                {
                    let mut rpc_pending = self.rpc_pending.lock().await;
                    if let Some(response_tx) = rpc_pending.remove(&result.correlation_id) {
                        let response = if result.success {
                            result
                                .payload
                                .map(|p| {
                                    if let InvocationResultPayload::Rpc(v) = p {
                                        Ok(v)
                                    } else {
                                        Err(crate::rpc::errors::RpcError::internal_error(
                                            "Wrong payload type".to_string(),
                                        ))
                                    }
                                })
                                .unwrap_or_else(|| {
                                    Err(crate::rpc::errors::RpcError::internal_error(
                                        "No RPC payload".to_string(),
                                    ))
                                })
                        } else {
                            Err(crate::rpc::errors::RpcError::internal_error(
                                result.error.unwrap_or_else(|| "Unknown error".to_string()),
                            ))
                        };
                        let _ = response_tx.send(response);
                        return Ok(());
                    }
                }
                // Fallback: pending_invocations (invoke_cli/invoke_rpc/invoke_timer).
                // A late result whose correlation was already dropped is ignored.
                self.complete_pending_invocation(result).await;
                return Ok(());
            }
            ModuleMessage::Invocation(_) => {
                warn!("Received Invocation from module (unexpected; node sends to module)");
                return Ok(());
            }
            ModuleMessage::Log(log_msg) => {
                // Forward log message to node's logging system
                use crate::module::ipc::protocol::LogLevel;
                let module_id_str = log_msg.module_id.clone();
                let message_str = log_msg.message.clone();
                // Use tracing macros without target parameter (tracing will use default target)
                // The module_id is included in the log message for identification
                match log_msg.level {
                    LogLevel::Trace => {
                        tracing::trace!(
                            module_id = %module_id_str,
                            "{}",
                            message_str
                        );
                    }
                    LogLevel::Debug => {
                        tracing::debug!(
                            module_id = %module_id_str,
                            "{}",
                            message_str
                        );
                    }
                    LogLevel::Info => {
                        tracing::info!(
                            module_id = %module_id_str,
                            "{}",
                            message_str
                        );
                    }
                    LogLevel::Warn => {
                        tracing::warn!(
                            module_id = %module_id_str,
                            "{}",
                            message_str
                        );
                    }
                    LogLevel::Error => {
                        tracing::error!(
                            module_id = %module_id_str,
                            "{}",
                            message_str
                        );
                    }
                }
                // Log messages don't require a response
                return Ok(());
            }
        }

        Ok(())
    }

    /// Process a request from a module
    async fn process_request<A: NodeAPI + Send + Sync>(
        &self,
        request: &RequestMessage,
        node_api: Arc<A>,
        module_id: &str,
    ) -> Result<ResponseMessage, ModuleError> {
        use crate::module::ipc::protocol::{RequestPayload, ResponsePayload};

        struct ClearModuleId;
        impl Drop for ClearModuleId {
            fn drop(&mut self) {
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
            }
        }
        crate::module::api::node_api::NodeApiImpl::set_current_module_id(module_id.to_string());
        let _module_id_guard = ClearModuleId;

        match &request.payload {
            RequestPayload::Handshake { .. } => {
                // Handshake is handled at connection level
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::HandshakeAck {
                        node_version: env!("CARGO_PKG_VERSION").to_string(),
                    },
                ))
            }
            RequestPayload::GetBlock { hash } => {
                let block = node_api.get_block(hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Block(block),
                ))
            }
            RequestPayload::GetBlockHeader { hash } => {
                let header = node_api.get_block_header(hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockHeader(header),
                ))
            }
            RequestPayload::GetTransaction { hash } => {
                let tx = node_api.get_transaction(hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Transaction(tx),
                ))
            }
            RequestPayload::HasTransaction { hash } => {
                let exists = node_api.has_transaction(hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(exists),
                ))
            }
            RequestPayload::GetChainTip => {
                let tip = node_api.get_chain_tip().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Hash(tip),
                ))
            }
            RequestPayload::GetBlockHeight => {
                let height = node_api.get_block_height().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::U64(height),
                ))
            }
            RequestPayload::GetUtxo { outpoint } => {
                let utxo = node_api.get_utxo(outpoint).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Utxo(utxo),
                ))
            }
            RequestPayload::SubscribeEvents { event_types } => {
                // Register module subscriptions with event manager
                if let Some(_event_mgr) = &self.event_manager {
                    // Get module ID from connection (would need to pass it through)
                    // For now, we'll handle this in handle_message where we have connection
                    // This will be implemented properly when we integrate event manager
                    debug!("Module subscribing to events: {:?}", event_types);
                }
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::SubscribeAck,
                ))
            }
            // Mempool API
            RequestPayload::GetMempoolTransactions => {
                let txs = node_api.get_mempool_transactions().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MempoolTransactions(txs),
                ))
            }
            RequestPayload::GetMempoolTransaction { tx_hash } => {
                let tx = node_api.get_mempool_transaction(tx_hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MempoolTransaction(tx),
                ))
            }
            RequestPayload::GetMempoolSize => {
                let size = node_api.get_mempool_size().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MempoolSize(size),
                ))
            }
            // Network API
            RequestPayload::GetNetworkStats => {
                let stats = node_api.get_network_stats().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::NetworkStats(stats),
                ))
            }
            RequestPayload::GetNetworkPeers => {
                let peers = node_api.get_network_peers().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::NetworkPeers(peers),
                ))
            }
            // Chain API
            RequestPayload::GetChainInfo => {
                let info = node_api.get_chain_info().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ChainInfo(info),
                ))
            }
            RequestPayload::GetBlockByHeight { height } => {
                let block = node_api.get_block_by_height(*height).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockByHeight(block),
                ))
            }
            RequestPayload::GetBlockAndWitnesses { hash } => {
                let pair = node_api.get_block_and_witnesses(hash).await?;
                let (block, witnesses) = match pair {
                    Some((block, witnesses)) => (Some(block), witnesses),
                    None => (None, Vec::new()),
                };
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockAndWitnesses { block, witnesses },
                ))
            }
            // Lightning API
            RequestPayload::GetLightningNodeUrl => {
                let url = node_api.get_lightning_node_url().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::LightningNodeUrl(url),
                ))
            }
            RequestPayload::GetLightningInfo => {
                let info = node_api.get_lightning_info().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::LightningInfo(info),
                ))
            }
            // Payment API
            RequestPayload::GetPaymentState { payment_id } => {
                let state = node_api.get_payment_state(payment_id).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::PaymentState(state),
                ))
            }
            // Additional Mempool API
            RequestPayload::CheckTransactionInMempool { tx_hash } => {
                let exists = node_api.check_transaction_in_mempool(tx_hash).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::CheckTransactionInMempool(exists),
                ))
            }
            RequestPayload::GetFeeEstimate { target_blocks } => {
                match node_api.get_fee_estimate(*target_blocks).await {
                    Ok(fee_rate) => Ok(ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::FeeEstimate(fee_rate),
                    )),
                    Err(err) => Ok(ResponseMessage::error(
                        request.correlation_id,
                        err.to_string(),
                    )),
                }
            }
            // Filesystem API
            RequestPayload::ReadFile { path } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let data = node_api.read_file(path.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                match data {
                    Ok(data) => Ok(ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::FileData(data),
                    )),
                    Err(e) => Ok(ResponseMessage::error(
                        request.correlation_id,
                        e.to_string(),
                    )),
                }
            }
            RequestPayload::WriteFile { path, data } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let result = node_api.write_file(path.clone(), data.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                result?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::DeleteFile { path } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let result = node_api.delete_file(path.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                result?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::ListDirectory { path } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let entries = node_api.list_directory(path.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                let entries = entries?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::DirectoryListing(entries),
                ))
            }
            RequestPayload::CreateDirectory { path } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let result = node_api.create_directory(path.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                result?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::GetFileMetadata { path } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let metadata = node_api.get_file_metadata(path.clone()).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                let metadata = metadata?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::FileMetadata(metadata),
                ))
            }
            // Module RPC Endpoint Registration
            RequestPayload::RegisterRpcEndpoint {
                method,
                description,
            } => {
                node_api
                    .register_rpc_endpoint(method.clone(), description.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::RpcEndpointRegistered,
                ))
            }
            RequestPayload::UnregisterRpcEndpoint { method } => {
                node_api.unregister_rpc_endpoint(method).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::RpcEndpointUnregistered,
                ))
            }
            RequestPayload::RegisterCoreRpcOverride {
                method,
                description,
            } => {
                node_api
                    .register_core_rpc_override(method.clone(), description.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::CoreRpcOverrideRegistered,
                ))
            }
            RequestPayload::UnregisterCoreRpcOverride { method } => {
                node_api.unregister_core_rpc_override(method).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::CoreRpcOverrideUnregistered,
                ))
            }
            // Timers and Scheduled Tasks
            RequestPayload::RegisterTimer { .. }
            | RequestPayload::CancelTimer { .. }
            | RequestPayload::ScheduleTask { .. } => Ok(ResponseMessage::error(
                request.correlation_id,
                crate::module::timers::types::LEGACY_TIMER_UNSUPPORTED.to_string(),
            )),
            // Metrics and Telemetry
            RequestPayload::ReportMetric { metric } => {
                node_api.report_metric(metric.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MetricReported,
                ))
            }
            RequestPayload::GetModuleMetrics { module_id } => {
                let metrics = node_api.get_module_metrics(module_id).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleMetrics(metrics),
                ))
            }
            RequestPayload::GetAllMetrics => {
                let metrics = node_api.get_all_metrics().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::AllMetrics(metrics),
                ))
            }
            // Network Integration
            RequestPayload::SendMeshPacketToPeer {
                peer_addr,
                packet_data,
            } => {
                node_api
                    .send_mesh_packet_to_peer(peer_addr.clone(), packet_data.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::SendStratumV2MessageToPeer {
                peer_addr,
                message_data,
            } => {
                node_api
                    .send_peer_transport_payload(peer_addr.clone(), message_data.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            // Mining API
            RequestPayload::GetBlockTemplate {
                rules,
                coinbase_script,
                coinbase_address,
            } => {
                let template = node_api
                    .get_block_template(
                        rules.clone(),
                        coinbase_script.clone(),
                        coinbase_address.clone(),
                    )
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockTemplate(template),
                ))
            }
            RequestPayload::GetBlockTemplateWithOutputs {
                rules,
                coinbase_script,
                coinbase_outputs,
            } => {
                let template = node_api
                    .get_block_template_with_outputs(
                        rules.clone(),
                        coinbase_script.clone(),
                        coinbase_outputs.clone(),
                    )
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockTemplate(template),
                ))
            }
            RequestPayload::GetBlockTemplateDeclared {
                rules,
                coinbase_script,
                coinbase_outputs,
                declared_txids,
            } => {
                let template = node_api
                    .get_block_template_declared(
                        rules.clone(),
                        coinbase_script.clone(),
                        coinbase_outputs.clone(),
                        declared_txids.clone(),
                    )
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockTemplate(template),
                ))
            }
            RequestPayload::SubmitMempoolTransaction { tx, witnesses } => {
                let added = node_api
                    .submit_mempool_transaction(tx.clone(), witnesses.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MempoolTransactionSubmitted(added),
                ))
            }
            RequestPayload::SubmitBlock { block } => {
                let result = node_api.submit_block(block.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::SubmitBlockResult(result),
                ))
            }
            RequestPayload::QueueReceivedBlock { block_bytes } => {
                node_api
                    .queue_received_block_bytes(block_bytes.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ReceivedBlockQueued,
                ))
            }
            RequestPayload::SetAdvertisedServiceBits { bits } => {
                crate::module::api::node_api::NodeApiImpl::set_current_module_id(
                    module_id.to_string(),
                );
                let result = node_api.set_advertised_service_bits(*bits).await;
                crate::module::api::node_api::NodeApiImpl::clear_current_module_id();
                match result {
                    Ok(()) => Ok(ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::AdvertisedServiceBitsSet,
                    )),
                    Err(e) => Ok(ResponseMessage::error(
                        request.correlation_id,
                        e.to_string(),
                    )),
                }
            }
            RequestPayload::MergeBlockServeDenylist { block_hashes } => {
                node_api
                    .merge_block_serve_denylist(block_hashes.as_slice())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockServeDenylistMerged,
                ))
            }
            RequestPayload::GetBlockServeDenylistSnapshot => {
                let s = node_api.get_block_serve_denylist_snapshot().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::BlockServeDenylistSnapshot(s),
                ))
            }
            RequestPayload::ClearBlockServeDenylist => {
                node_api.clear_block_serve_denylist().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::ReplaceBlockServeDenylist { block_hashes } => {
                node_api
                    .replace_block_serve_denylist(block_hashes.as_slice())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::MergeTxServeDenylist { tx_hashes } => {
                node_api
                    .merge_tx_serve_denylist(tx_hashes.as_slice())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TxServeDenylistMerged,
                ))
            }
            RequestPayload::GetTxServeDenylistSnapshot => {
                let s = node_api.get_tx_serve_denylist_snapshot().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TxServeDenylistSnapshot(s),
                ))
            }
            RequestPayload::ClearTxServeDenylist => {
                node_api.clear_tx_serve_denylist().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::ReplaceTxServeDenylist { tx_hashes } => {
                node_api
                    .replace_tx_serve_denylist(tx_hashes.as_slice())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::GetSyncStatus => {
                let s = node_api.get_sync_status().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::NodeSyncStatus(s),
                ))
            }
            RequestPayload::BanPeer {
                peer_addr,
                ban_duration_seconds,
            } => {
                node_api
                    .ban_peer(peer_addr.as_str(), *ban_duration_seconds)
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::SetBlockServeMaintenanceMode { enabled } => {
                node_api.set_block_serve_maintenance_mode(*enabled).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::GetTxMetadata { txid } => {
                let metadata = node_api.get_tx_metadata(txid).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TxMetadata(metadata),
                ))
            }
            RequestPayload::GetMempoolFeeRateSummary => {
                let summary = node_api.get_mempool_fee_rate_summary().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::MempoolFeeRateSummary(summary),
                ))
            }
            RequestPayload::DiscoverModules => {
                let modules = node_api.discover_modules().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleList(modules),
                ))
            }
            RequestPayload::GetModuleInfo { module_id } => {
                let info = node_api.get_module_info(module_id).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleInfo(info),
                ))
            }
            RequestPayload::IsModuleAvailable { module_id } => {
                let available = node_api.is_module_available(module_id).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleAvailable(available),
                ))
            }
            RequestPayload::PublishEvent {
                event_type,
                payload,
            } => match node_api.publish_event(*event_type, payload.clone()).await {
                Ok(()) => Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::EventPublished,
                )),
                Err(err) => Ok(ResponseMessage::error(
                    request.correlation_id,
                    err.to_string(),
                )),
            },
            RequestPayload::SubscribeModuleEventNames { .. } => Ok(ResponseMessage::success(
                request.correlation_id,
                ResponsePayload::SubscribeAck,
            )),
            RequestPayload::CallModule {
                target_module_id,
                method,
                params,
            } => {
                let response = node_api
                    .call_module(target_module_id.as_deref(), method.as_str(), params.clone())
                    .await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleApiResponse(response),
                ))
            }
            RequestPayload::RegisterModuleApi { .. } => {
                // Handled at connection level in hub with module_id context
                Err(ModuleError::OperationError(
                    "RegisterModuleApi must be handled via API hub".to_string(),
                ))
            }
            RequestPayload::UnregisterModuleApi => {
                node_api.unregister_module_api().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleApiUnregistered,
                ))
            }
            RequestPayload::GetModuleHealth { module_id } => {
                let health = node_api.get_module_health(module_id).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::ModuleHealth(health),
                ))
            }
            RequestPayload::GetAllModuleHealth => {
                let health = node_api.get_all_module_health().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::AllModuleHealth(health),
                ))
            }
            RequestPayload::ReportModuleHealth { health } => {
                node_api.report_module_health(health.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::HealthReported,
                ))
            }
            RequestPayload::TimerCreate { spec, mode } => {
                let info = node_api.timer_create(spec.clone(), *mode).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TimerCreated(info),
                ))
            }
            RequestPayload::TimerCancel { target } => {
                node_api.timer_cancel(target.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::TimerPause { target } => {
                node_api.timer_pause(target.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::TimerResume { target } => {
                node_api.timer_resume(target.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::TimerTrigger { target } => {
                node_api.timer_trigger(target.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::Bool(true),
                ))
            }
            RequestPayload::TimerGet { target } => {
                let info = node_api.timer_get(target.clone()).await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TimerInfo(info),
                ))
            }
            RequestPayload::RegisterFeeEstimators { specs } => {
                match node_api
                    .register_fee_estimators(module_id, specs.clone())
                    .await
                {
                    Ok(registration) => Ok(ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::FeeEstimatorsRegistered(registration),
                    )),
                    Err(err) => Ok(ResponseMessage::error(
                        request.correlation_id,
                        err.to_string(),
                    )),
                }
            }
            RequestPayload::SubmitFeeTable { table } => {
                match node_api.submit_fee_table(module_id, table.clone()).await {
                    Ok(generation) => Ok(ResponseMessage::success(
                        request.correlation_id,
                        ResponsePayload::FeeTableAccepted { generation },
                    )),
                    Err(err) => Ok(ResponseMessage::error(
                        request.correlation_id,
                        err.to_string(),
                    )),
                }
            }
            RequestPayload::GetFeeTable {
                view,
                mode,
                targets,
            } => match node_api.get_fee_table(view, mode, targets).await {
                Ok(table) => Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::NativeFeeTable(table),
                )),
                Err(err) => Ok(ResponseMessage::error(
                    request.correlation_id,
                    err.to_string(),
                )),
            },
            RequestPayload::TimerList => {
                let list = node_api.timer_list().await?;
                Ok(ResponseMessage::success(
                    request.correlation_id,
                    ResponsePayload::TimerList(list),
                ))
            }
            _ => Ok(ResponseMessage::error(
                request.correlation_id,
                format!("Unimplemented request payload: {:?}", request.payload),
            )),
        }
    }
}

/// Delivers timer fires through the module IPC connection.
pub struct IpcTimerTransport {
    handle: ModuleIpcHandle,
}

impl IpcTimerTransport {
    pub fn new(handle: ModuleIpcHandle) -> Self {
        Self { handle }
    }
}

#[async_trait::async_trait]
impl crate::module::timers::service::TimerTransport for IpcTimerTransport {
    async fn invoke(
        &self,
        module_id: &str,
        fire: crate::module::timers::types::TimerFire,
        ack_timeout: std::time::Duration,
    ) -> Result<
        crate::module::timers::types::TimerAck,
        crate::module::timers::service::TransportError,
    > {
        use crate::module::timers::service::TransportError;
        match self.handle.invoke_timer(module_id, fire, ack_timeout).await {
            Ok(ack) => Ok(ack),
            Err(ModuleError::Timeout) => Err(TransportError::Timeout),
            Err(ModuleError::OperationError(msg))
                if msg.contains("closed") || msg.contains("not connected") =>
            {
                Err(TransportError::Closed)
            }
            Err(err) => Err(TransportError::Failed(err.to_string())),
        }
    }
}

/// Derive Windows named pipe name from socket path (e.g. "hello.sock" -> "\\.\pipe\blvm-hello")
#[cfg(windows)]
fn path_to_pipe_name(path: &Path) -> String {
    let stem = path
        .file_stem()
        .and_then(|s| s.to_str())
        .unwrap_or("blvm-module");
    let safe = stem.replace(|c: char| !c.is_alphanumeric(), "-");
    format!(r"\\.\pipe\blvm-{}", safe)
}

#[cfg(test)]
mod timer_invoke_tests {
    use super::*;
    use crate::module::timers::types::{Clock, TimerFire, TimerId};
    use std::time::Duration;

    fn sample_fire() -> TimerFire {
        TimerFire {
            timer_id: TimerId([1; 16]),
            key: None,
            payload: Vec::new(),
            seq: 1,
            attempt: 0,
            clock: Clock::Monotonic,
            scheduled_unix_ms: 0,
            planned_unix_ms: 0,
            fired_unix_ms: 0,
            lateness_ms: 0,
            missed: 0,
            coalesced: 0,
            skipped_overlap: 0,
            clock_step_ms: 0,
            tip_height: None,
            tip_hash: None,
            tip_mtp: None,
            is_last: false,
        }
    }

    #[tokio::test]
    async fn timeout_removes_the_pending_invocation() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let handle = ModuleIpcHandle {
            outgoing_tx_by_module: Arc::new(tokio::sync::RwLock::new(HashMap::from([(
                "fees".to_string(),
                tx,
            )]))),
            pending_invocations: Arc::new(tokio::sync::Mutex::new(HashMap::new())),
            next_invocation_id: Arc::new(AtomicU64::new(1)),
            cli_registry: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
        };
        tokio::spawn(async move { while rx.recv().await.is_some() {} });
        let err = handle
            .invoke_timer("fees", sample_fire(), Duration::from_millis(20))
            .await
            .unwrap_err();
        assert!(matches!(err, ModuleError::Timeout));
        assert_eq!(handle.pending_invocation_count().await, 0);
    }

    #[tokio::test]
    async fn late_result_for_a_missing_correlation_is_ignored() {
        let server = ModuleIpcServer::new("unused-timer.sock");
        let handle = server.handle();
        server
            .complete_pending_invocation(InvocationResultMessage {
                correlation_id: 99,
                success: true,
                payload: Some(InvocationResultPayload::Timer(
                    crate::module::timers::types::TimerAck::Continue,
                )),
                error: None,
            })
            .await;
        assert_eq!(handle.pending_invocation_count().await, 0);
    }
}