blvm-node 0.1.90

//! Module manager for orchestrating all modules
//!
//! Handles module lifecycle, runtime loading/unloading/reloading, and coordination.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tokio::sync::{Mutex, mpsc};
use tokio::task::JoinHandle;
use tracing::{debug, error, info, warn};

use crate::module::api::events::EventManager;
use crate::module::api::hub::ModuleApiHub;
#[cfg(unix)]
use crate::module::ipc::server::ModuleIpcServer;
#[cfg(all(unix, feature = "wasm-modules"))]
use crate::module::ipc::server::WasmInvoker;
use crate::module::loader::ModuleLoader;
use crate::module::process::{
    monitor::ModuleProcessMonitor,
    spawner::{ModuleProcess, ModuleProcessSpawner},
};
use crate::module::registry::{ModuleDependencies, ModuleDiscovery};
use crate::module::security::permissions::PermissionSet;
use crate::module::traits::{ModuleContext, ModuleError, ModuleMetadata, ModuleState};
#[cfg(feature = "wasm-modules")]
use crate::module::wasm::{WasmModuleInstance, WasmModuleLoader};
use crate::utils::{MODULE_RELOAD_CLEANUP_DELAY, new_request_id};

/// Module manager coordinates all loaded modules
pub struct ModuleManager {
    /// Process spawner
    spawner: ModuleProcessSpawner,
    /// Active modules (name -> process)
    modules: Arc<Mutex<HashMap<String, ManagedModule>>>,
    /// IPC server handle
    ipc_server_handle: Option<JoinHandle<Result<(), ModuleError>>>,
    /// IPC server reference (for getmoduleclispecs, etc.)
    #[cfg(unix)]
    ipc_server: Option<Arc<tokio::sync::Mutex<ModuleIpcServer>>>,
    /// Lock-free handle to IPC server state — use this for cross-task calls to
    /// avoid deadlocking with the server's own Mutex (held for its entire lifetime).
    #[cfg(unix)]
    ipc_handle: Option<crate::module::ipc::server::ModuleIpcHandle>,
    /// Crash notification receiver (mutable so it can be moved to handler)
    crash_rx: Option<mpsc::UnboundedReceiver<(String, ModuleError)>>,
    /// Crash notification sender
    crash_tx: mpsc::UnboundedSender<(String, ModuleError)>,
    /// Base directory for module binaries
    modules_dir: PathBuf,
    /// Node config overrides per module ([modules.<name>] from config.toml). Merged over module config.
    module_config_overrides: std::sync::RwLock<HashMap<String, HashMap<String, String>>>,
    /// Default database backend for modules (inherited from node when not set per-module).
    /// Values: "redb", "rocksdb", "sled", "heed3".
    default_database_backend: std::sync::RwLock<Option<String>>,
    /// Event manager for module event subscriptions
    event_manager: Arc<EventManager>,
    /// API hub for request routing
    api_hub: Option<Arc<tokio::sync::Mutex<crate::module::api::hub::ModuleApiHub>>>,
    /// Module registry for fetching modules via P2P
    module_registry: Option<Arc<crate::module::registry::client::ModuleRegistry>>,
    /// WASM loader (injected by binary; e.g. from blvm-sdk)
    #[cfg(feature = "wasm-modules")]
    wasm_loader: Option<Arc<dyn WasmModuleLoader>>,
    /// RPC server reference — used to clean up module-registered endpoints on unload.
    rpc_server: Option<Arc<crate::rpc::server::RpcServer>>,
    /// If non-empty, only these module names are auto-loaded (from config `enabled_modules` keys).
    /// Values are version constraints (`0.1.*`, exact semver, or `"*"` for unpinned).
    /// An empty map means "load everything discovered".
    enabled_modules: HashMap<String, String>,
    /// Never auto-load or bootstrap these manifest names (from config `disabled_modules`).
    disabled_modules: Vec<String>,
    /// HTTP URL for the first-party module registry JSON (e.g. blvm/registry/modules.json on GitHub).
    /// When set, `auto_load_modules` will bootstrap-download any `enabled_modules` not found locally.
    registry_url: Option<String>,
    /// When false, a module with no signatures is not loaded.
    allow_unsigned_modules: bool,
    /// Node IPC socket path (where node listens; modules connect here)
    node_socket_path: Option<PathBuf>,
    /// Node P2P listen address injected into module spawn env (`MODULE_CONFIG_NODE_P2P_LISTEN_*`).
    node_p2p_listen: std::sync::RwLock<Option<std::net::SocketAddr>>,
    /// Shared timer service. Attached to the IPC server when `start` runs.
    timer_service: Option<crate::module::timers::TimerService>,
    /// Module metrics. Cleared when a module unloads or crashes.
    metrics_manager: Option<Arc<crate::module::metrics::MetricsManager>>,
    /// Operator HTTP allow-list and proxy. Applied when the API hub starts.
    http_policy: crate::module::http::HttpPolicy,
    /// Inbound listeners. Unload and crash drop them.
    service_host: crate::module::services::ServiceHost,
    /// `[services.<name>]` from node config.
    service_configs: HashMap<String, crate::config::HostedServiceConfig>,
    /// Drops a module's fee estimators on unload and crash.
    fee_estimator: Option<Arc<crate::node::fee_estimator::FeeEstimator>>,
}

/// Managed module instance
struct ManagedModule {
    /// Module metadata
    metadata: ModuleMetadata,
    /// Module process (native subprocess; None for WASM modules)
    process: Option<Arc<tokio::sync::Mutex<ModuleProcess>>>,
    /// Module state
    state: ModuleState,
    /// Monitoring handle (native only)
    monitor_handle: Option<JoinHandle<()>>,
    /// Process ID for tracking (native only)
    process_id: Option<u32>,
    /// Handshake id `{name}_{32 hex}`. Metrics are stored under this id.
    module_id: String,
    /// WASM instance (in-process; None for native modules)
    #[cfg(feature = "wasm-modules")]
    wasm_instance: Option<Arc<tokio::sync::Mutex<Arc<dyn WasmModuleInstance>>>>,
}

/// In-process WASM invoker for the IPC server.
#[cfg(all(unix, feature = "wasm-modules"))]
struct ManagerWasmInvoker {
    modules: Arc<Mutex<HashMap<String, ManagedModule>>>,
}

#[cfg(all(unix, feature = "wasm-modules"))]
#[async_trait::async_trait]
impl WasmInvoker for ManagerWasmInvoker {
    async fn invoke_cli(
        &self,
        module_name: &str,
        subcommand: &str,
        args: Vec<String>,
    ) -> Result<crate::module::ipc::protocol::InvocationResultPayload, ModuleError> {
        let wasm_guard = {
            let modules = self.modules.lock().await;
            let managed = modules.get(module_name).ok_or_else(|| {
                ModuleError::OperationError(format!("No WASM module '{}' is loaded", module_name))
            })?;
            #[cfg(feature = "wasm-modules")]
            match &managed.wasm_instance {
                Some(inst) => inst.clone(),
                None => {
                    return Err(ModuleError::OperationError(format!(
                        "Module '{}' is not a WASM module",
                        module_name
                    )));
                }
            }
        };
        let instance_guard = wasm_guard.lock().await;
        let result =
            tokio::task::block_in_place(|| (**instance_guard).invoke_cli(subcommand, args));
        result
    }

    async fn invoke_rpc(
        &self,
        module_name: &str,
        method: &str,
        params: serde_json::Value,
    ) -> Result<serde_json::Value, ModuleError> {
        let wasm_guard = {
            let modules = self.modules.lock().await;
            let managed = modules.get(module_name).ok_or_else(|| {
                ModuleError::OperationError(format!("No WASM module '{}' is loaded", module_name))
            })?;
            #[cfg(feature = "wasm-modules")]
            match &managed.wasm_instance {
                Some(inst) => inst.clone(),
                None => {
                    return Err(ModuleError::OperationError(format!(
                        "Module '{}' is not a WASM module",
                        module_name
                    )));
                }
            }
        };
        let instance_guard = wasm_guard.lock().await;
        let result = tokio::task::block_in_place(|| (**instance_guard).invoke_rpc(method, params));
        result
    }

    async fn on_event(&self, module_name: &str, event_json: &str) -> Result<String, ModuleError> {
        let wasm_guard = {
            let modules = self.modules.lock().await;
            let managed = modules.get(module_name).ok_or_else(|| {
                ModuleError::OperationError(format!("No WASM module '{}' is loaded", module_name))
            })?;
            match &managed.wasm_instance {
                Some(inst) => inst.clone(),
                None => {
                    return Err(ModuleError::OperationError(format!(
                        "Module '{}' is not a WASM module",
                        module_name
                    )));
                }
            }
        };
        let instance_guard = wasm_guard.lock().await;
        tokio::task::block_in_place(|| instance_guard.on_event(event_json))
    }
}

#[cfg(feature = "wasm-modules")]
fn wasm_host_node_call_blocking(
    hub: &Arc<tokio::sync::Mutex<ModuleApiHub>>,
    module_name: &str,
    method: &str,
    args: &str,
) -> Result<String, String> {
    let hub = Arc::clone(hub);
    let module_name = module_name.to_string();
    let method = method.to_string();
    let args = args.to_string();
    let runtime = tokio::runtime::Handle::current();
    let (tx, rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let result = runtime.block_on(async move {
            let hub = hub.lock().await;
            match method.as_str() {
                "get_block_height" => hub
                    .wasm_host_get_block_height(&module_name)
                    .await
                    .map(|height| height.to_string())
                    .map_err(|e| e.to_string()),
                "get_block" => {
                    let hash = wasm_block_hash_arg(&args)?;
                    hub.wasm_host_get_block(&module_name, &hash)
                        .await
                        .map_err(|e| e.to_string())
                        .and_then(|block| serde_json::to_string(&block).map_err(|e| e.to_string()))
                }
                _ => Err("host_node_call method not available".to_string()),
            }
        });
        let _ = tx.send(result);
    });
    rx.recv().unwrap_or_else(|e| Err(e.to_string()))
}

#[cfg(feature = "wasm-modules")]
fn wasm_block_hash_arg(args: &str) -> Result<crate::Hash, String> {
    let parsed: serde_json::Value =
        serde_json::from_str(args).map_err(|e| format!("get_block args: {e}"))?;
    let hex_hash = parsed
        .get("hash")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "get_block args require {\"hash\":\"<64 hex>\"}".to_string())?;
    let bytes = hex::decode(hex_hash).map_err(|e| format!("get_block hash: {e}"))?;
    let hash: crate::Hash = bytes
        .try_into()
        .map_err(|_| "get_block hash must be 32 bytes".to_string())?;
    Ok(hash)
}

fn safe_module_dir_name(name: &str) -> bool {
    !name.is_empty()
        && !name.contains('/')
        && !name.contains('\\')
        && !name.contains("..")
        && name
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
}

impl ModuleManager {
    /// Create a new module manager
    pub fn new<P: AsRef<Path>>(modules_dir: P, data_dir: P, socket_dir: P) -> Self {
        Self::with_config(modules_dir, data_dir, socket_dir, None)
    }

    /// Create a new module manager with resource limits configuration
    pub fn with_config<P: AsRef<Path>>(
        modules_dir: P,
        data_dir: P,
        socket_dir: P,
        resource_limits_config: Option<&crate::config::ModuleResourceLimitsConfig>,
    ) -> Self {
        let (crash_tx, crash_rx) = mpsc::unbounded_channel();

        Self {
            spawner: ModuleProcessSpawner::with_config(
                &modules_dir,
                &data_dir,
                &socket_dir,
                resource_limits_config,
            ),
            modules: Arc::new(Mutex::new(HashMap::new())),
            ipc_server_handle: None,
            #[cfg(unix)]
            ipc_server: None,
            #[cfg(unix)]
            ipc_handle: None,
            crash_rx: Some(crash_rx),
            crash_tx,
            modules_dir: modules_dir.as_ref().to_path_buf(),
            module_config_overrides: std::sync::RwLock::new(HashMap::new()),
            default_database_backend: std::sync::RwLock::new(None),
            event_manager: Arc::new(EventManager::new()),
            api_hub: None,
            module_registry: None,
            #[cfg(feature = "wasm-modules")]
            wasm_loader: None,
            node_socket_path: None,
            rpc_server: None,
            enabled_modules: HashMap::new(),
            disabled_modules: Vec::new(),
            registry_url: None,
            allow_unsigned_modules: false,
            node_p2p_listen: std::sync::RwLock::new(None),
            timer_service: None,
            metrics_manager: None,
            http_policy: crate::module::http::HttpPolicy::default(),
            service_host: crate::module::services::ServiceHost::new(),
            service_configs: HashMap::new(),
            fee_estimator: None,
        }
    }

    /// The node's fee estimator. Unload and crash drop that module's names.
    pub fn set_fee_estimator(&mut self, estimator: Arc<crate::node::fee_estimator::FeeEstimator>) {
        self.fee_estimator = Some(estimator);
    }

    /// Install the node timer service. Call this before `start`.
    pub fn set_timer_service(&mut self, service: crate::module::timers::TimerService) {
        self.timer_service = Some(service);
    }

    /// Operator HTTP allow-list and proxy. Call this before `start`.
    pub fn set_http_policy(&mut self, policy: crate::module::http::HttpPolicy) {
        self.http_policy = policy;
    }

    /// `[services.<name>]` entries. Call this before modules load.
    pub fn set_service_configs(
        &mut self,
        configs: HashMap<String, crate::config::HostedServiceConfig>,
    ) {
        self.service_configs = configs;
    }

    /// Listener table shared with the API hub.
    pub fn service_host(&self) -> crate::module::services::ServiceHost {
        self.service_host.clone()
    }

    /// Rename the first existing `replaces` data directory onto this module
    /// when the new directory does not exist yet.
    pub fn adopt_replaced_data(
        &self,
        manifest: &crate::module::registry::manifest::ModuleManifest,
    ) -> Result<(), ModuleError> {
        let new_dir = self.spawner.data_dir.join(&manifest.name);
        if new_dir.exists() {
            return Ok(());
        }
        for old_name in &manifest.replaces {
            if !safe_module_dir_name(old_name) || old_name == &manifest.name {
                return Err(ModuleError::InvalidManifest(format!(
                    "refusing to rename module data from '{old_name}'"
                )));
            }
            let old_dir = self.spawner.data_dir.join(old_name);
            if !old_dir.is_dir() {
                continue;
            }
            if let Some(parent) = new_dir.parent() {
                std::fs::create_dir_all(parent).map_err(|err| {
                    ModuleError::InitializationError(format!(
                        "failed to create module data parent: {err}"
                    ))
                })?;
            }
            std::fs::rename(&old_dir, &new_dir).map_err(|err| {
                ModuleError::InitializationError(format!(
                    "failed to rename module data '{old_name}' to '{}': {err}",
                    manifest.name
                ))
            })?;
            info!(
                from = %old_name,
                to = %manifest.name,
                "renamed module data directory"
            );
            return Ok(());
        }
        Ok(())
    }

    /// A loaded module whose manifest `provides_apis` contains the contract
    /// satisfies `requires_apis`. The name of that module is not part of the match.
    pub async fn check_requires_apis(
        &self,
        manifest: &crate::module::registry::manifest::ModuleManifest,
    ) -> Result<(), ModuleError> {
        for api in &manifest.requires_apis {
            if !self.service_host.provides_api(api).await {
                return Err(ModuleError::DependencyMissing(format!(
                    "required api '{api}' is not provided by a loaded module"
                )));
            }
        }
        Ok(())
    }

    /// Remember `provides_apis` after a successful load.
    pub async fn record_declaration(
        &self,
        manifest: &crate::module::registry::manifest::ModuleManifest,
    ) {
        self.service_host
            .record_provides(&manifest.name, manifest.provides_apis.clone())
            .await;
    }

    /// Bind `[services.*]` listeners this manifest declares and the operator enabled.
    pub async fn bind_declared_services(
        &self,
        manifest: &crate::module::registry::manifest::ModuleManifest,
    ) -> Result<(), ModuleError> {
        let requests = self.collect_bind_requests(manifest)?;
        if requests.is_empty() {
            return Ok(());
        }
        let allowed = manifest.capabilities.iter().any(|cap| {
            crate::module::security::permissions::parse_permission_string(cap)
                == Some(crate::module::security::permissions::Permission::HostService)
        });
        if !allowed {
            return Err(ModuleError::OperationError(format!(
                "module {} declares a hosted service but does not have host_service",
                manifest.name
            )));
        }
        #[cfg(unix)]
        {
            let Some(handle) = self.ipc_handle.clone() else {
                return Err(ModuleError::OperationError(
                    "hosted service requires the module IPC server".to_string(),
                ));
            };
            let module_id = {
                let modules = self.modules.lock().await;
                modules
                    .get(&manifest.name)
                    .map(|module| module.module_id.clone())
            };
            let Some(module_id) = module_id else {
                return Err(ModuleError::OperationError(
                    "hosted service bind requires the module to be loaded".to_string(),
                ));
            };
            let handler = Arc::new(crate::module::services::IpcHostedModule { handle, module_id });
            self.service_host
                .bind(&manifest.name, requests, handler)
                .await
                .map_err(ModuleError::OperationError)?;
            Ok(())
        }
        #[cfg(not(unix))]
        {
            let _ = requests;
            Err(ModuleError::OperationError(
                "hosted services require the Unix IPC server".to_string(),
            ))
        }
    }

    fn collect_bind_requests(
        &self,
        manifest: &crate::module::registry::manifest::ModuleManifest,
    ) -> Result<Vec<crate::module::services::BindRequest>, ModuleError> {
        let mut requests = Vec::new();
        for svc in &manifest.services {
            let Some(cfg) = self.service_configs.get(&svc.name) else {
                continue;
            };
            if cfg.module != manifest.name {
                continue;
            }
            let Some(listen) = cfg.listen.as_deref().filter(|listen| !listen.is_empty()) else {
                continue;
            };
            if cfg.module.is_empty() {
                return Err(ModuleError::OperationError(format!(
                    "services.{}.module is required when listen is set",
                    svc.name
                )));
            }
            let listen = listen.parse().map_err(|err| {
                ModuleError::OperationError(format!(
                    "services.{} listen '{listen}' is not a socket address: {err}",
                    svc.name
                ))
            })?;
            let kind = match svc.kind.as_str() {
                "http" => crate::module::services::ServiceKind::Http,
                "stream" => crate::module::services::ServiceKind::Stream,
                other => {
                    return Err(ModuleError::OperationError(format!(
                        "service {} kind '{other}' is not http or stream",
                        svc.name
                    )));
                }
            };
            requests.push(crate::module::services::BindRequest {
                name: svc.name.clone(),
                kind,
                listen,
                max_connections: cfg
                    .max_connections
                    .unwrap_or(crate::module::services::DEFAULT_MAX_CONNECTIONS),
                tls_cert: cfg.tls_cert.clone(),
            });
        }
        Ok(requests)
    }

    /// Record a stopped module so `unload_module` runs listener cleanup.
    /// Does not spawn a process.
    pub async fn remember_stopped_module(&mut self, name: &str) {
        let mut modules = self.modules.lock().await;
        modules.insert(
            name.to_string(),
            ManagedModule {
                metadata: ModuleMetadata {
                    name: name.to_string(),
                    version: "0.0.0".to_string(),
                    description: String::new(),
                    author: String::new(),
                    capabilities: Vec::new(),
                    rpc_overrides: Vec::new(),
                    dependencies: HashMap::new(),
                    optional_dependencies: HashMap::new(),
                    entry_point: name.to_string(),
                },
                process: None,
                state: ModuleState::Stopped,
                monitor_handle: None,
                process_id: None,
                module_id: format!("{name}_00000000000000000000000000000000"),
                #[cfg(feature = "wasm-modules")]
                wasm_instance: None,
            },
        );
    }

    /// Install the module metrics manager. Unload and crash clear that module's series.
    pub fn set_metrics_manager(&mut self, manager: Arc<crate::module::metrics::MetricsManager>) {
        self.metrics_manager = Some(manager);
    }

    /// Set the WASM module loader (injected by binary; e.g. from blvm-sdk).
    #[cfg(feature = "wasm-modules")]
    pub fn set_wasm_loader(&mut self, loader: Arc<dyn WasmModuleLoader>) {
        self.wasm_loader = Some(loader);
    }

    /// Last `on_event` return string for a loaded WASM module.
    #[cfg(feature = "wasm-modules")]
    pub async fn wasm_last_on_event_result(&self, module_name: &str) -> Option<String> {
        let instance = {
            let modules = self.modules.lock().await;
            modules
                .get(module_name)
                .and_then(|managed| managed.wasm_instance.clone())?
        };
        let guard = instance.lock().await;
        guard.last_on_event_result()
    }

    /// Set node config overrides for modules ([modules.<name>] from config.toml).
    /// Merged over module's config.toml when loading. Node values take precedence.
    pub fn set_module_config_overrides(&self, overrides: HashMap<String, HashMap<String, String>>) {
        if let Ok(mut guard) = self.module_config_overrides.write() {
            *guard = overrides;
        }
    }

    /// Set node's P2P listen address for module subprocesses.
    ///
    /// Merged into spawn config (unless already overridden) as `node_p2p_listen_ip` and
    /// `node_p2p_listen_port`, exposed as `MODULE_CONFIG_NODE_P2P_LISTEN_*` env vars.
    /// Used by **`blvm-fibre`** with `udp_follow_node_tcp_plus_one` to match in-process FIBRE
    /// (`UDP port = TCP listen port + 1`).
    pub fn set_node_p2p_listen_for_modules(&self, addr: std::net::SocketAddr) {
        if let Ok(mut g) = self.node_p2p_listen.write() {
            *g = Some(addr);
            debug!(
                "Module spawn env: node P2P listen {} (for MODULE_CONFIG_NODE_P2P_LISTEN_*)",
                addr
            );
        }
    }

    fn apply_node_p2p_listen_to_module_config(&self, merged: &mut HashMap<String, String>) {
        let Ok(g) = self.node_p2p_listen.read() else {
            return;
        };
        let Some(addr) = *g else {
            return;
        };
        merged
            .entry("node_p2p_listen_ip".to_string())
            .or_insert_with(|| addr.ip().to_string());
        merged
            .entry("node_p2p_listen_port".to_string())
            .or_insert_with(|| addr.port().to_string());
    }

    /// Set default `database_backend` merged into module spawn env (`MODULE_CONFIG_DATABASE_BACKEND`).
    ///
    /// Set from node startup using
    /// [`module_subprocess_database_backend_preference`](crate::storage::database::module_subprocess_database_backend_preference)
    /// so it reflects `[storage]` + optional `[modules] module_database_backend`.
    pub fn set_default_database_backend(&self, backend: String) {
        if let Ok(mut guard) = self.default_database_backend.write() {
            *guard = Some(backend);
        }
    }

    /// Merge base config with node overrides for a module.
    fn merge_module_config(
        &self,
        module_name: &str,
        base: HashMap<String, String>,
    ) -> HashMap<String, String> {
        let mut merged = base;
        if let Ok(guard) = self.module_config_overrides.read() {
            if let Some(overrides) = guard.get(module_name) {
                for (k, v) in overrides {
                    merged.insert(k.clone(), v.clone());
                }
            }
        }
        // Inherit node's effective module DB backend when module doesn't specify one
        if !merged.contains_key("database_backend") {
            if let Ok(guard) = self.default_database_backend.read() {
                if let Some(ref backend) = *guard {
                    merged.insert("database_backend".to_string(), backend.clone());
                }
            }
        }
        merged
    }

    /// Set module registry for fetching modules via P2P
    pub fn with_module_registry(
        mut self,
        module_registry: Arc<crate::module::registry::client::ModuleRegistry>,
    ) -> Self {
        self.module_registry = Some(module_registry);
        self
    }

    /// Set module registry for fetching modules via P2P (mutable reference version)
    pub fn set_module_registry(
        &mut self,
        module_registry: Arc<crate::module::registry::client::ModuleRegistry>,
    ) {
        self.module_registry = Some(module_registry);
    }

    /// Get modules directory path (for discovery)
    pub fn modules_dir(&self) -> &Path {
        &self.modules_dir
    }

    /// Get IPC server reference (for getmoduleclispecs; Unix only)
    #[cfg(unix)]
    pub fn ipc_server(&self) -> Option<Arc<tokio::sync::Mutex<ModuleIpcServer>>> {
        self.ipc_server.clone()
    }

    /// Lock-free handle for cross-task module API invocations.
    #[cfg(unix)]
    pub fn ipc_handle(&self) -> Option<crate::module::ipc::server::ModuleIpcHandle> {
        self.ipc_handle.clone()
    }

    /// Attach the RPC server so `unload_module` can clean up registered endpoints/overrides.
    pub fn with_rpc_server(&mut self, rpc_server: Arc<crate::rpc::server::RpcServer>) {
        self.rpc_server = Some(rpc_server);
    }

    /// Set enabled modules and version constraints from config.
    /// If the map is non-empty, `auto_load_modules` only loads modules whose name appears here.
    /// An empty map (the default) means load every discovered module.
    pub fn set_enabled_modules(&mut self, enabled: HashMap<String, String>) {
        self.enabled_modules = enabled;
    }

    /// Allow loading a module that has no signatures. Default is refuse.
    pub fn set_allow_unsigned_modules(&mut self, allow: bool) {
        self.allow_unsigned_modules = allow;
    }

    pub fn allow_unsigned_modules(&self) -> bool {
        self.allow_unsigned_modules
    }

    /// Set module names to exclude from auto-load and registry bootstrap (opt-out).
    pub fn set_disabled_modules(&mut self, disabled: Vec<String>) {
        self.disabled_modules = disabled;
    }

    /// Set the HTTP URL for the first-party module registry JSON.
    /// When set, `auto_load_modules` will bootstrap-download any `enabled_modules` not found locally.
    pub fn set_registry_url(&mut self, url: String) {
        self.registry_url = Some(url);
    }

    /// HTTP URL for the module registry index (`modules.json`), if configured.
    pub fn registry_url(&self) -> Option<&str> {
        self.registry_url.as_deref()
    }

    /// Return a clone of the `api_hub` arc so callers can release the `ModuleManager` lock
    /// before making async inter-module calls.
    ///
    /// Pattern for callers that hold the `ModuleManager` mutex:
    /// ```ignore
    /// let hub_arc = {
    ///     let manager = mgr.lock().await;
    ///     manager.api_hub_arc()
    /// }; // mgr lock released here
    /// if let Some(hub) = hub_arc {
    ///     let node_api = hub.lock().await.node_api();
    ///     let resp = node_api.call_module(Some(target), method, params).await?;
    /// }
    /// ```
    pub fn api_hub_arc(
        &self,
    ) -> Option<Arc<tokio::sync::Mutex<crate::module::api::hub::ModuleApiHub>>> {
        self.api_hub.clone()
    }

    /// Start the module manager
    pub async fn start<
        P: AsRef<Path>,
        A: crate::module::traits::NodeAPI + Send + Sync + 'static,
    >(
        &mut self,
        socket_path: P,
        node_api: Arc<A>,
    ) -> Result<(), ModuleError> {
        info!("Starting module manager");

        // Create API hub
        let api_hub = Arc::new(tokio::sync::Mutex::new(ModuleApiHub::new(Arc::clone(
            &node_api,
        ))));
        {
            let mut hub = api_hub.lock().await;
            hub.set_http_policy(self.http_policy.clone());
            hub.set_service_host(self.service_host.clone());
        }
        self.api_hub = Some(Arc::clone(&api_hub));

        // Start IPC server in background task (Unix only: domain sockets)
        #[cfg(unix)]
        {
            let mut ipc_server = ModuleIpcServer::new(&socket_path)
                .with_event_manager(Arc::clone(&self.event_manager))
                .with_api_hub(Arc::clone(&api_hub));
            if let Some(service) = self.timer_service.clone() {
                ipc_server = ipc_server.with_timer_service(service);
            }
            #[cfg(all(unix, feature = "wasm-modules"))]
            if self.wasm_loader.is_some() {
                let invoker = Arc::new(ManagerWasmInvoker {
                    modules: Arc::clone(&self.modules),
                });
                ipc_server = ipc_server.with_wasm_invoker(invoker);
            }
            // Capture a lock-free handle before moving the server into Arc<Mutex>.
            let ipc_handle = ipc_server.handle();
            if let Some(service) = self.timer_service.clone() {
                service
                    .set_transport(Arc::new(
                        crate::module::ipc::server::IpcTimerTransport::new(ipc_handle.clone()),
                    ))
                    .await;
            }
            let ipc_server = Arc::new(tokio::sync::Mutex::new(ipc_server));
            self.ipc_server = Some(Arc::clone(&ipc_server));
            self.ipc_handle = Some(ipc_handle);
            self.node_socket_path = Some(socket_path.as_ref().to_path_buf());
            let node_api_clone = Arc::clone(&node_api);
            let server_handle =
                tokio::spawn(async move { ipc_server.lock().await.start(node_api_clone).await });
            self.ipc_server_handle = Some(server_handle);
        } // end #[cfg(unix)]

        // Start crash handler
        let modules = Arc::clone(&self.modules);
        let event_manager = Arc::clone(&self.event_manager);
        let api_hub = self.api_hub.clone();
        let timer_service = self.timer_service.clone();
        let metrics_manager = self.metrics_manager.clone();
        let service_host = self.service_host.clone();
        let fee_estimator = self.fee_estimator.clone();
        #[cfg(unix)]
        let ipc_server = self.ipc_server.clone();
        if let Some(mut crash_rx) = self.crash_rx.take() {
            tokio::spawn(async move {
                while let Some((module_name, error)) = crash_rx.recv().await {
                    warn!("Module {} crashed: {}", module_name, error);

                    // Get dependent modules before removing
                    let dependents: Vec<String> = {
                        let modules = modules.lock().await;
                        modules
                            .iter()
                            .filter_map(|(name, m)| {
                                if m.metadata.dependencies.contains_key(&module_name) {
                                    Some(name.clone())
                                } else {
                                    None
                                }
                            })
                            .collect()
                    };

                    // Remove crashed module
                    let crashed_module_id = {
                        let mut modules = modules.lock().await;
                        if let Some(mut managed) = modules.remove(&module_name) {
                            // Stop monitoring
                            if let Some(handle) = managed.monitor_handle.take() {
                                handle.abort();
                            }
                            managed.state = ModuleState::Error(error.to_string());
                            Some(managed.module_id)
                        } else {
                            None
                        }
                    };
                    if let Some(module_id) = crashed_module_id.as_deref() {
                        if let Some(metrics) = &metrics_manager {
                            metrics.clear_module_metrics(module_id).await;
                        }
                    }

                    // Clean up: loaded_modules, CLI spec, API hub (same as explicit unload)
                    event_manager.remove_loaded_module(&module_name).await;
                    if let Some(service) = &timer_service {
                        service.detach_module(&module_name, None).await;
                    }
                    #[cfg(unix)]
                    if let Some(ref ipc) = ipc_server {
                        let ipc = ipc.lock().await;
                        if let Some(module_id) = crashed_module_id.as_deref() {
                            ipc.unregister_cli_spec(module_id).await;
                        }
                        ipc.unregister_cli_spec_by_name(&module_name).await;
                    }
                    service_host.forget(&module_name).await;
                    if let Some(estimator) = &fee_estimator {
                        if let Some(module_id) = crashed_module_id.as_deref() {
                            estimator.drop_module_estimates(module_id);
                        }
                        estimator.drop_module_estimates_named(&module_name);
                    }
                    if let Some(ref hub) = api_hub {
                        let mut h = hub.lock().await;
                        if let Err(e) = h.clear_module_service_bits(&module_name).await {
                            warn!(
                                "Failed to clear service bits for crashed module {}: {}",
                                module_name, e
                            );
                        }
                        h.unregister_module(&module_name).await;
                    }

                    // Publish ModuleStateChanged and ModuleHealthChanged (Running -> Error)
                    use crate::module::ipc::protocol::EventPayload;
                    use crate::module::traits::{EventType, ModuleError};
                    let payload = EventPayload::ModuleStateChanged {
                        module_name: module_name.clone(),
                        old_state: "Running".to_string(),
                        new_state: "Error".to_string(),
                    };
                    if let Err(e) = event_manager
                        .publish_event(EventType::ModuleStateChanged, payload)
                        .await
                    {
                        warn!(
                            "Failed to publish ModuleStateChanged for crashed module: {}",
                            e
                        );
                    }
                    let payload = EventPayload::ModuleHealthChanged {
                        module_name: module_name.clone(),
                        old_health: "healthy".to_string(),
                        new_health: "unhealthy".to_string(),
                    };
                    if let Err(e) = event_manager
                        .publish_event(EventType::ModuleHealthChanged, payload)
                        .await
                    {
                        warn!(
                            "Failed to publish ModuleHealthChanged for crashed module: {}",
                            e
                        );
                    }

                    // Publish ModuleCrashed event (specific to abnormal exit)
                    let error_msg = match &error {
                        ModuleError::ModuleCrashed(msg) => msg.clone(),
                        _ => error.to_string(),
                    };
                    let payload = EventPayload::ModuleCrashed {
                        module_name: module_name.clone(),
                        error: error_msg,
                    };
                    if let Err(e) = event_manager
                        .publish_event(EventType::ModuleCrashed, payload)
                        .await
                    {
                        warn!(
                            "Failed to publish ModuleCrashed event for {}: {}",
                            module_name, e
                        );
                    }

                    // Publish ModuleUnloaded event (crashed modules are effectively unloaded)
                    let payload = EventPayload::ModuleUnloaded {
                        module_name: module_name.clone(),
                        version: String::new(), // Version unknown for crashed modules
                    };
                    if let Err(e) = event_manager
                        .publish_event(EventType::ModuleUnloaded, payload)
                        .await
                    {
                        warn!(
                            "Failed to publish ModuleUnloaded event for crashed module: {}",
                            e
                        );
                    }

                    // Unload dependent modules with hard dependencies
                    // Note: We can't use self.unload_module() here since we're in a spawned task
                    // Instead, we'll just remove them and let the system handle it
                    if !dependents.is_empty() {
                        warn!(
                            "Unloading {} dependent module(s) due to crashed dependency '{}'",
                            dependents.len(),
                            module_name
                        );
                        let removed: Vec<String> = {
                            let mut modules = modules.lock().await;
                            dependents
                                .into_iter()
                                .filter_map(|dependent| {
                                    if let Some(mut managed) = modules.remove(&dependent) {
                                        if let Some(handle) = managed.monitor_handle.take() {
                                            handle.abort();
                                        }
                                        managed.state = ModuleState::Error(format!(
                                            "Dependency '{module_name}' crashed"
                                        ));
                                        warn!(
                                        "Dependent module '{}' unloaded due to crashed dependency",
                                        dependent
                                    );
                                        Some(dependent)
                                    } else {
                                        None
                                    }
                                })
                                .collect()
                        };
                        for dependent in removed {
                            event_manager.remove_loaded_module(&dependent).await;
                            if let Some(service) = &timer_service {
                                service.detach_module(&dependent, None).await;
                            }
                            #[cfg(unix)]
                            if let Some(ref ipc) = ipc_server {
                                ipc.lock()
                                    .await
                                    .unregister_cli_spec_by_name(&dependent)
                                    .await;
                            }
                            if let Some(ref hub) = api_hub {
                                let mut h = hub.lock().await;
                                h.unregister_module(&dependent).await;
                            }
                        }
                    }
                }
            });
        } else {
            warn!("Module crash receiver already taken; crash events will not be handled");
        }

        info!("Module manager started");
        Ok(())
    }

    /// Load a module at runtime
    pub async fn load_module(
        &mut self,
        module_name: &str,
        binary_path: &Path,
        metadata: ModuleMetadata,
        config: HashMap<String, String>,
    ) -> Result<(), ModuleError> {
        info!("Loading module: {}", module_name);

        let mut modules = self.modules.lock().await;

        // Check if module already loaded
        if modules.contains_key(module_name) {
            return Err(ModuleError::OperationError(format!(
                "Module {module_name} is already loaded"
            )));
        }

        // Validate rpc_overrides against the allowlist BEFORE spawning the process.
        for method in &metadata.rpc_overrides {
            if !crate::rpc::methods::OVERRIDABLE_CORE_RPC_METHODS.contains(&method.as_str()) {
                return Err(ModuleError::OperationError(format!(
                    "Module '{module_name}' declares rpc_override '{method}' which is not in OVERRIDABLE_CORE_RPC_METHODS"
                )));
            }
        }

        // Validate dependencies BEFORE spawning process (hard dependency enforcement)
        for (dep_name, dep_version_req) in &metadata.dependencies {
            // Check if dependency is loaded
            let dep_module = modules.get(dep_name).ok_or_else(|| {
                ModuleError::DependencyMissing(format!(
                    "Required dependency '{dep_name}' not loaded (required by '{module_name}')"
                ))
            })?;

            // Validate version constraint (basic semver checking)
            if !Self::check_version_constraint(&dep_module.metadata.version, dep_version_req) {
                return Err(ModuleError::DependencyMissing(format!(
                    "Dependency '{}' version '{}' does not satisfy requirement '{}' (required by '{}')",
                    dep_name, dep_module.metadata.version, dep_version_req, module_name
                )));
            }

            // Check if dependency is in a valid state (Running or Initialized)
            if dep_module.state != ModuleState::Running
                && dep_module.state != ModuleState::Initializing
            {
                return Err(ModuleError::OperationError(format!(
                    "Dependency '{}' is not in a valid state (state: {:?}, required by '{}')",
                    dep_name, dep_module.state, module_name
                )));
            }
        }

        // Create module context
        let module_id = format!("{module_name}_{}", new_request_id());
        let socket_path = self
            .node_socket_path
            .as_ref()
            .cloned()
            .unwrap_or_else(|| self.spawner.socket_dir.join(format!("{module_name}.sock")));
        let data_dir = self.spawner.data_dir.join(module_name);

        // Ensure module data directory exists
        std::fs::create_dir_all(&data_dir).map_err(|e| {
            ModuleError::InitializationError(format!("Failed to create module data directory: {e}"))
        })?;

        let mut merged_config = self.merge_module_config(module_name, config);
        self.apply_node_p2p_listen_to_module_config(&mut merged_config);
        merged_config.insert("version".to_string(), metadata.version.clone());
        let config_for_wasm = merged_config.clone();
        let context = ModuleContext::new(
            module_id.clone(),
            socket_path.to_string_lossy().to_string(),
            data_dir.to_string_lossy().to_string(),
            merged_config,
        );

        use std::sync::Arc;

        let is_wasm = binary_path
            .extension()
            .map(|e| e == "wasm")
            .unwrap_or(false);

        // WASM path: load in-process via injected loader
        if is_wasm {
            #[cfg(feature = "wasm-modules")]
            {
                let loader = self.wasm_loader.as_ref().ok_or_else(|| {
                    ModuleError::InitializationError(
                        "WASM loader not configured. Pass WasmModuleLoader when creating the node."
                            .to_string(),
                    )
                })?;
                let data_dir = data_dir.to_path_buf();
                let instance = loader
                    .load(binary_path, &data_dir, config_for_wasm)
                    .map_err(|e| {
                        ModuleError::InitializationError(format!("Failed to load WASM module: {e}"))
                    })?;
                let cli_spec =
                    instance
                        .cli_spec()
                        .unwrap_or_else(|| crate::module::ipc::protocol::CliSpec {
                            version: 1,
                            name: module_name.to_string(),
                            about: None,
                            subcommands: vec![],
                        });

                if let Some(ref api_hub) = self.api_hub {
                    let permissions = Self::parse_permissions_from_metadata(&metadata);
                    let mut hub_guard = api_hub.lock().await;
                    hub_guard.register_module_permissions(module_name.to_string(), permissions);
                    drop(hub_guard);
                    let hub = Arc::clone(api_hub);
                    let call_name = module_name.to_string();
                    instance.set_host_node_call(std::sync::Arc::new(move |method, args| {
                        wasm_host_node_call_blocking(&hub, &call_name, method, args)
                    }));
                }

                let event_types = instance.event_types().map_err(|e| {
                    ModuleError::InitializationError(format!(
                        "WASM module {module_name} event_types: {e}"
                    ))
                })?;
                let (event_tx, mut event_rx) = mpsc::channel(32);
                self.event_manager
                    .subscribe_module(module_name.to_string(), event_types, event_tx)
                    .await
                    .map_err(|e| {
                        ModuleError::InitializationError(format!(
                            "WASM module {module_name} subscribe: {e}"
                        ))
                    })?;
                let wasm_instance = Arc::new(tokio::sync::Mutex::new(instance));
                let task_instance = Arc::clone(&wasm_instance);
                let task_name = module_name.to_string();
                let monitor_handle = tokio::spawn(async move {
                    while let Some(msg) = event_rx.recv().await {
                        let json = match serde_json::to_string(&msg) {
                            Ok(json) => json,
                            Err(e) => {
                                warn!("WASM module {task_name} event encode: {e}");
                                continue;
                            }
                        };
                        let guard = task_instance.lock().await;
                        if let Err(e) = tokio::task::block_in_place(|| guard.on_event(&json)) {
                            warn!("WASM module {task_name} on_event: {e}");
                        }
                    }
                });

                let module_version = metadata.version.clone();
                let managed = ManagedModule {
                    metadata,
                    process: None,
                    state: ModuleState::Running,
                    monitor_handle: Some(monitor_handle),
                    process_id: None,
                    module_id: module_id.clone(),
                    wasm_instance: Some(wasm_instance),
                };
                modules.insert(module_name.to_string(), managed);
                drop(modules);
                #[cfg(unix)]
                if let Some(ref ipc) = self.ipc_handle {
                    ipc.register_cli_spec(module_name.to_string(), cli_spec)
                        .await;
                }
                #[cfg(not(unix))]
                let _ = cli_spec;

                {
                    use crate::module::ipc::protocol::EventPayload;
                    use crate::module::traits::EventType;
                    let payload = EventPayload::ModuleInstalled {
                        module_name: module_name.to_string(),
                        version: module_version.clone(),
                    };
                    if let Err(e) = self
                        .event_manager
                        .publish_event(EventType::ModuleInstalled, payload)
                        .await
                    {
                        warn!("Failed to publish ModuleInstalled: {}", e);
                    }
                }

                let loaded_modules = self.event_manager.loaded_modules();
                let mut loaded = loaded_modules.lock().await;
                let timestamp = crate::utils::current_timestamp();
                loaded.insert(module_name.to_string(), (module_version, timestamp));
                info!("Module {} loaded successfully (WASM)", module_name);
                return Ok(());
            }
            #[cfg(not(feature = "wasm-modules"))]
            {
                return Err(ModuleError::InitializationError(
                    "WASM modules require --features wasm-modules".to_string(),
                ));
            }
        }

        // Native path: spawn subprocess
        let process = self
            .spawner
            .spawn(module_name, binary_path, context)
            .await?;
        let process_id = process.id();
        let shared_process = Arc::new(tokio::sync::Mutex::new(process));
        let monitor = ModuleProcessMonitor::new(self.crash_tx.clone());
        let module_name_clone = module_name.to_string();
        let shared_process_for_monitor = Arc::clone(&shared_process);
        let resource_sandbox = self.spawner.process_sandbox();
        let monitor_handle = tokio::spawn(async move {
            if let Err(e) = monitor
                .monitor_module_shared(
                    module_name_clone.clone(),
                    shared_process_for_monitor,
                    resource_sandbox,
                )
                .await
            {
                error!("Module {} monitor error: {}", module_name_clone, e);
            }
        });

        if let Some(ref api_hub) = self.api_hub {
            let permissions = Self::parse_permissions_from_metadata(&metadata);
            let mut hub_guard = api_hub.lock().await;
            hub_guard.register_module_permissions(module_name.to_string(), permissions);
        }

        let module_version = metadata.version.clone();
        let managed = ManagedModule {
            metadata,
            process: Some(shared_process),
            state: ModuleState::Running,
            monitor_handle: Some(monitor_handle),
            process_id,
            module_id,
            #[cfg(feature = "wasm-modules")]
            wasm_instance: None,
        };

        modules.insert(module_name.to_string(), managed);

        info!("Module {} loaded successfully", module_name);

        // Publish ModuleInstalled for registry/marketplace sync
        {
            use crate::module::ipc::protocol::EventPayload;
            use crate::module::traits::EventType;
            let payload = EventPayload::ModuleInstalled {
                module_name: module_name.to_string(),
                version: module_version.clone(),
            };
            if let Err(e) = self
                .event_manager
                .publish_event(EventType::ModuleInstalled, payload)
                .await
            {
                warn!("Failed to publish ModuleInstalled: {}", e);
            }
        }

        // Record module as loaded (for sending to newly subscribing modules)
        // ModuleLoaded event will be published AFTER the module subscribes (in subscribe_module)
        // This ensures consistency: ModuleLoaded only happens after startup is complete
        {
            let loaded_modules = self.event_manager.loaded_modules();
            let mut loaded = loaded_modules.lock().await;
            let timestamp = crate::utils::current_timestamp();
            loaded.insert(module_name.to_string(), (module_version.clone(), timestamp));
        }

        // Note: ModuleLoaded event is NOT published here
        // It will be published when the module subscribes to events (after startup is complete)
        // This ensures:
        // 1. ModuleLoaded only happens after module is fully ready (subscribed)
        // 2. Hotloaded modules receive ModuleLoaded for all already-loaded modules via subscribe_module()
        // 3. Consistent event ordering: subscription -> ModuleLoaded events

        Ok(())
    }

    /// Unload a module (stop and remove)
    pub async fn unload_module(&mut self, module_name: &str) -> Result<(), ModuleError> {
        info!("Unloading module: {}", module_name);

        // Get list of dependent modules (before we drop the lock)
        let dependents = self.get_dependent_modules(module_name).await;

        let mut modules = self.modules.lock().await;

        if let Some(mut managed) = modules.remove(module_name) {
            let metrics_id = managed.module_id.clone();
            // Publish ModuleStateChanged and ModuleHealthChanged before teardown
            let old_state = format!("{:?}", managed.state);
            let old_health = match &managed.state {
                ModuleState::Running | ModuleState::Initializing => "healthy",
                ModuleState::Stopping => "degraded",
                ModuleState::Stopped | ModuleState::Error(_) => "unhealthy",
            };
            drop(modules);
            self.service_host.forget(module_name).await;
            if let Some(estimator) = &self.fee_estimator {
                estimator.drop_module_estimates(&metrics_id);
                estimator.drop_module_estimates_named(module_name);
            }
            if let Some(metrics) = &self.metrics_manager {
                metrics.clear_module_metrics(&metrics_id).await;
            }
            {
                use crate::module::ipc::protocol::EventPayload;
                use crate::module::traits::EventType;
                let payload = EventPayload::ModuleStateChanged {
                    module_name: module_name.to_string(),
                    old_state: old_state.clone(),
                    new_state: "Stopping".to_string(),
                };
                if let Err(e) = self
                    .event_manager
                    .publish_event(EventType::ModuleStateChanged, payload)
                    .await
                {
                    warn!("Failed to publish ModuleStateChanged: {}", e);
                }
                let payload = EventPayload::ModuleHealthChanged {
                    module_name: module_name.to_string(),
                    old_health: old_health.to_string(),
                    new_health: "degraded".to_string(),
                };
                if let Err(e) = self
                    .event_manager
                    .publish_event(EventType::ModuleHealthChanged, payload)
                    .await
                {
                    warn!("Failed to publish ModuleHealthChanged: {}", e);
                }
            }

            // Stop monitoring (native process monitor, or the WASM event task)
            if let Some(handle) = managed.monitor_handle.take() {
                handle.abort();
            }
            #[cfg(feature = "wasm-modules")]
            if managed.wasm_instance.is_some() {
                if let Err(e) = self.event_manager.unsubscribe_module(module_name).await {
                    warn!(
                        "Failed to unsubscribe WASM module {} from events: {}",
                        module_name, e
                    );
                }
            }

            // Kill process if we have a reference
            if let Some(shared_process) = managed.process.take() {
                let mut process_guard = shared_process.lock().await;
                process_guard.kill().await?;
            } else if let Some(pid) = managed.process_id {
                warn!(
                    "Module {} has pid {} but no process handle; not signaling an unreaped pid",
                    module_name, pid
                );
            }

            info!("Module {} unloaded", module_name);

            // Unregister CLI spec (WASM modules register on load; native remove on disconnect, but we clean up here too)
            #[cfg(unix)]
            if let Some(ref ipc) = self.ipc_server {
                let ipc = ipc.lock().await;
                ipc.unregister_cli_spec(&metrics_id).await;
                ipc.unregister_cli_spec_by_name(module_name).await;
            }
            if let Some(service) = &self.timer_service {
                service.detach_module(module_name, None).await;
            }

            // Unregister all RPC extension endpoints and core overrides owned by this module.
            if let Some(ref rpc_server) = self.rpc_server {
                rpc_server.unregister_all_for_module(module_name).await;
            }

            // Remove from loaded_modules (event manager)
            self.event_manager.remove_loaded_module(module_name).await;

            // Unregister permissions and rate limiters (API hub)
            if let Some(ref api_hub) = self.api_hub {
                let mut hub = api_hub.lock().await;
                if let Err(e) = hub.clear_module_service_bits(module_name).await {
                    warn!(
                        "Failed to clear service bits for unloaded module {}: {}",
                        module_name, e
                    );
                }
                hub.unregister_module(module_name).await;
            }

            // Publish ModuleUnloaded event for dependent modules to react
            let version = managed.metadata.version.clone();
            {
                use crate::module::ipc::protocol::EventPayload;
                use crate::module::traits::EventType;
                let payload = EventPayload::ModuleUnloaded {
                    module_name: module_name.to_string(),
                    version: version.clone(),
                };
                if let Err(e) = self
                    .event_manager
                    .publish_event(EventType::ModuleUnloaded, payload)
                    .await
                {
                    warn!("Failed to publish ModuleUnloaded event: {}", e);
                }
                let payload = EventPayload::ModuleRemoved {
                    module_name: module_name.to_string(),
                    version,
                };
                if let Err(e) = self
                    .event_manager
                    .publish_event(EventType::ModuleRemoved, payload)
                    .await
                {
                    warn!("Failed to publish ModuleRemoved: {}", e);
                }
            }

            // Automatically unload dependent modules (if they have hard dependencies)
            for dependent in dependents {
                // Check if it's a hard dependency (required) or soft (optional)
                let is_required = {
                    let modules = self.modules.lock().await;
                    modules
                        .get(&dependent)
                        .map(|m| m.metadata.dependencies.contains_key(module_name))
                        .unwrap_or(false)
                };

                if is_required {
                    warn!(
                        "Unloading dependent module '{}' due to required dependency '{}' unloading",
                        dependent, module_name
                    );
                    if let Err(e) = Box::pin(self.unload_module(&dependent)).await {
                        error!("Failed to unload dependent module '{}': {}", dependent, e);
                    }
                } else {
                    debug!(
                        "Dependent module '{}' has optional dependency on '{}', leaving it running",
                        dependent, module_name
                    );
                }
            }

            Ok(())
        } else {
            Err(ModuleError::ModuleNotFound(module_name.to_string()))
        }
    }

    /// Reload a module (unload and load again)
    ///
    /// This gracefully handles dependent modules:
    /// 1. Tracks which dependent modules were unloaded (due to hard dependencies)
    /// 2. Reloads the module
    /// 3. Automatically reloads dependent modules that were unloaded
    /// 4. Handles version changes and validates dependency constraints
    /// 5. Publishes ModuleReloaded event
    ///
    /// # Errors
    /// - Returns error if the module cannot be reloaded
    /// - Dependent modules that fail to reload are logged but don't fail the operation
    pub async fn reload_module(
        &mut self,
        module_name: &str,
        binary_path: &Path,
        metadata: ModuleMetadata,
        config: HashMap<String, String>,
    ) -> Result<(), ModuleError> {
        info!("Reloading module: {}", module_name);

        // Get old version and dependent modules BEFORE unloading
        let (old_version, dependents_to_reload) = {
            let modules = self.modules.lock().await;
            let old_version = modules
                .get(module_name)
                .map(|m| m.metadata.version.clone())
                .unwrap_or_else(|| "unknown".to_string());

            // Get list of dependent modules that have hard dependencies
            // Use discovery to get proper binary paths and metadata
            let discovery =
                crate::module::registry::discovery::ModuleDiscovery::new(&self.modules_dir);
            let dependents: Vec<String> = modules
                .iter()
                .filter_map(|(name, m)| {
                    // Check if this module has a hard dependency on the module being reloaded
                    if m.metadata.dependencies.contains_key(module_name) {
                        Some(name.clone())
                    } else {
                        None
                    }
                })
                .collect();

            (old_version, dependents)
        };

        // Unload the module (this will also unload dependents with hard dependencies)
        let unload_result = self.unload_module(module_name).await;
        if let Err(e) = unload_result {
            warn!("Error unloading module {} for reload: {}", module_name, e);
            // Continue anyway - might be partially unloaded
        }

        // Small delay to ensure cleanup
        tokio::time::sleep(MODULE_RELOAD_CLEANUP_DELAY).await;

        // Reload the module
        let reload_result = self
            .load_module(module_name, binary_path, metadata.clone(), config.clone())
            .await;

        // Publish ModuleReloaded and ModuleUpdated events
        use crate::module::ipc::protocol::EventPayload;
        use crate::module::traits::EventType;
        let new_version = metadata.version.clone();
        let payload = EventPayload::ModuleReloaded {
            module_name: module_name.to_string(),
            old_version: old_version.clone(),
            new_version: new_version.clone(),
        };
        if let Err(e) = self
            .event_manager
            .publish_event(EventType::ModuleReloaded, payload)
            .await
        {
            warn!("Failed to publish ModuleReloaded event: {}", e);
        }
        if reload_result.is_ok() {
            let payload = EventPayload::ModuleUpdated {
                module_name: module_name.to_string(),
                old_version: old_version.clone(),
                new_version: new_version.clone(),
            };
            if let Err(e) = self
                .event_manager
                .publish_event(EventType::ModuleUpdated, payload)
                .await
            {
                warn!("Failed to publish ModuleUpdated event: {}", e);
            }
        }

        // If reload succeeded, reload dependent modules using discovery
        if reload_result.is_ok() && !dependents_to_reload.is_empty() {
            info!(
                "Reloading {} dependent module(s) after {} reload",
                dependents_to_reload.len(),
                module_name
            );

            let discovery =
                crate::module::registry::discovery::ModuleDiscovery::new(&self.modules_dir);

            for dependent_name in dependents_to_reload {
                // Discover the dependent module to get proper binary path and metadata
                let discovered = match discovery.discover_module(&dependent_name) {
                    Ok(d) => d,
                    Err(e) => {
                        error!(
                            "Cannot reload dependent module '{}': discovery failed: {}",
                            dependent_name, e
                        );
                        continue; // Skip this dependent
                    }
                };

                // Validate that the new version still satisfies the dependent's requirements
                let version_req = discovered.manifest.dependencies.get(module_name);
                if let Some(req) = version_req {
                    if !Self::check_version_constraint(&metadata.version, req) {
                        error!(
                            "Cannot reload dependent module '{}': new version '{}' of '{}' does not satisfy requirement '{}'",
                            dependent_name, metadata.version, module_name, req
                        );
                        continue; // Skip this dependent
                    }
                }

                // Check if binary path exists
                if !discovered.binary_path.exists() {
                    warn!(
                        "Cannot reload dependent module '{}': binary not found at {}",
                        dependent_name,
                        discovered.binary_path.display()
                    );
                    continue; // Skip this dependent
                }

                // Load config if it exists
                let config_path = discovered.directory.join("config.toml");
                let dependent_config = if config_path.exists() {
                    // Try to load config from TOML file
                    match std::fs::read_to_string(&config_path) {
                        Ok(contents) => {
                            // Parse TOML config
                            match toml::from_str::<toml::Value>(&contents) {
                                Ok(toml_value) => {
                                    // Convert TOML value to HashMap<String, String>
                                    // This handles nested structures by flattening keys
                                    fn flatten_toml_value(
                                        value: &toml::Value,
                                        prefix: &str,
                                        result: &mut HashMap<String, String>,
                                    ) {
                                        match value {
                                            toml::Value::String(s) => {
                                                result.insert(
                                                    if prefix.is_empty() {
                                                        "value".to_string()
                                                    } else {
                                                        prefix.to_string()
                                                    },
                                                    s.clone(),
                                                );
                                            }
                                            toml::Value::Integer(i) => {
                                                result.insert(
                                                    if prefix.is_empty() {
                                                        "value".to_string()
                                                    } else {
                                                        prefix.to_string()
                                                    },
                                                    i.to_string(),
                                                );
                                            }
                                            toml::Value::Float(f) => {
                                                result.insert(
                                                    if prefix.is_empty() {
                                                        "value".to_string()
                                                    } else {
                                                        prefix.to_string()
                                                    },
                                                    f.to_string(),
                                                );
                                            }
                                            toml::Value::Boolean(b) => {
                                                result.insert(
                                                    if prefix.is_empty() {
                                                        "value".to_string()
                                                    } else {
                                                        prefix.to_string()
                                                    },
                                                    b.to_string(),
                                                );
                                            }
                                            toml::Value::Table(table) => {
                                                for (key, val) in table {
                                                    let new_prefix = if prefix.is_empty() {
                                                        key.clone()
                                                    } else {
                                                        format!("{prefix}.{key}")
                                                    };
                                                    flatten_toml_value(val, &new_prefix, result);
                                                }
                                            }
                                            toml::Value::Array(arr) => {
                                                for (idx, val) in arr.iter().enumerate() {
                                                    let new_prefix = format!("{prefix}[{idx}]");
                                                    flatten_toml_value(val, &new_prefix, result);
                                                }
                                            }
                                            toml::Value::Datetime(dt) => {
                                                result.insert(
                                                    if prefix.is_empty() {
                                                        "value".to_string()
                                                    } else {
                                                        prefix.to_string()
                                                    },
                                                    dt.to_string(),
                                                );
                                            }
                                        }
                                    }
                                    let mut config_map = HashMap::new();
                                    flatten_toml_value(&toml_value, "", &mut config_map);
                                    config_map
                                }
                                Err(e) => {
                                    warn!(
                                        "Failed to parse TOML config from {}: {}",
                                        config_path.display(),
                                        e
                                    );
                                    HashMap::new()
                                }
                            }
                        }
                        Err(e) => {
                            warn!(
                                "Failed to read TOML config from {}: {}",
                                config_path.display(),
                                e
                            );
                            HashMap::new()
                        }
                    }
                } else {
                    HashMap::new()
                };

                // Reload the dependent module
                info!("Reloading dependent module: {}", dependent_name);
                let dependent_metadata = discovered.manifest.to_metadata();
                match self
                    .load_module(
                        &dependent_name,
                        &discovered.binary_path,
                        dependent_metadata,
                        dependent_config,
                    )
                    .await
                {
                    Ok(()) => {
                        info!("Successfully reloaded dependent module: {}", dependent_name);
                    }
                    Err(e) => {
                        error!(
                            "Failed to reload dependent module '{}': {}",
                            dependent_name, e
                        );
                        // Continue with other dependents - don't fail the whole operation
                    }
                }
            }
        }

        reload_result
    }

    /// Get list of loaded modules
    pub async fn list_modules(&self) -> Vec<String> {
        let modules = self.modules.lock().await;
        modules.keys().cloned().collect()
    }

    /// Get module state
    pub async fn get_module_state(&self, module_name: &str) -> Option<ModuleState> {
        let modules = self.modules.lock().await;
        modules.get(module_name).map(|m| m.state.clone())
    }

    /// Get module metadata
    pub async fn get_module_metadata(&self, module_name: &str) -> Option<ModuleMetadata> {
        let modules = self.modules.lock().await;
        modules.get(module_name).map(|m| m.metadata.clone())
    }

    /// Get all module information for discovery
    pub async fn get_all_module_info(&self) -> Vec<(String, ModuleMetadata, ModuleState)> {
        let modules = self.modules.lock().await;
        modules
            .iter()
            .map(|(name, m)| (name.clone(), m.metadata.clone(), m.state.clone()))
            .collect()
    }

    /// Validate that all required dependencies for a module are available and running
    /// Only validates required (hard) dependencies - optional dependencies are checked at runtime
    pub async fn validate_module_dependencies(&self, module_name: &str) -> Result<(), ModuleError> {
        let modules = self.modules.lock().await;
        let module = modules.get(module_name).ok_or_else(|| {
            ModuleError::OperationError(format!("Module {module_name} not found"))
        })?;

        // Check each required (hard) dependency
        for (dep_name, dep_version_req) in &module.metadata.dependencies {
            // Check if dependency is loaded
            let dep_module = modules.get(dep_name).ok_or_else(|| {
                ModuleError::OperationError(format!(
                    "Required dependency '{dep_name}' not loaded (required by '{module_name}')"
                ))
            })?;

            // Validate version constraint
            if !Self::check_version_constraint(&dep_module.metadata.version, dep_version_req) {
                return Err(ModuleError::DependencyMissing(format!(
                    "Dependency '{}' version '{}' does not satisfy requirement '{}' (required by '{}')",
                    dep_name, dep_module.metadata.version, dep_version_req, module_name
                )));
            }

            // Check if dependency is in a valid state (Running or Initialized)
            if dep_module.state != ModuleState::Running
                && dep_module.state != ModuleState::Initializing
            {
                return Err(ModuleError::OperationError(format!(
                    "Required dependency '{}' is not in a valid state (state: {:?}, required by '{}')",
                    dep_name, dep_module.state, module_name
                )));
            }
        }

        // Optional dependencies are not validated here - they're checked at runtime when needed

        Ok(())
    }

    /// Get list of modules that depend on a given module (required or optional)
    pub async fn get_dependent_modules(&self, module_name: &str) -> Vec<String> {
        let modules = self.modules.lock().await;
        let mut dependents = Vec::new();

        for (name, module) in modules.iter() {
            // Check both required and optional dependencies
            if module.metadata.dependencies.contains_key(module_name)
                || module
                    .metadata
                    .optional_dependencies
                    .contains_key(module_name)
            {
                dependents.push(name.clone());
            }
        }

        dependents
    }

    /// Apply `disabled_modules`, `enabled_modules` allowlist, and version constraints.
    pub(crate) fn filter_discovered_for_auto_load(
        &self,
        mut discovered_modules: Vec<crate::module::registry::discovery::DiscoveredModule>,
    ) -> Vec<crate::module::registry::discovery::DiscoveredModule> {
        if !self.disabled_modules.is_empty() {
            let before = discovered_modules.len();
            discovered_modules.retain(|m| !self.disabled_modules.contains(&m.manifest.name));
            let after = discovered_modules.len();
            if before != after {
                info!(
                    "disabled_modules filter: kept {}/{} discovered modules ({} skipped)",
                    after,
                    before,
                    before - after
                );
            }
        }

        if !self.enabled_modules.is_empty() {
            let before = discovered_modules.len();
            discovered_modules.retain(|m| self.enabled_modules.contains_key(&m.manifest.name));
            let after = discovered_modules.len();
            if before != after {
                info!(
                    "enabled_modules filter: kept {}/{} discovered modules ({} skipped)",
                    after,
                    before,
                    before - after
                );
            }
        }

        if !self.enabled_modules.is_empty() {
            let before = discovered_modules.len();
            discovered_modules.retain(|m| {
                let Some(constraint) = self.enabled_modules.get(&m.manifest.name) else {
                    return false;
                };
                if crate::module::version_constraint::matches_version_constraint(
                    &m.manifest.version,
                    constraint,
                ) {
                    true
                } else {
                    warn!(
                        "Module '{}' version {} does not match constraint '{}'; will re-bootstrap",
                        m.manifest.name, m.manifest.version, constraint
                    );
                    false
                }
            });
            let after = discovered_modules.len();
            if before != after {
                info!(
                    "version constraint filter: kept {}/{} discovered modules ({} queued for re-bootstrap)",
                    after,
                    before,
                    before - after
                );
            }
        }

        discovered_modules
    }

    /// Auto-discover and load all modules
    pub async fn auto_load_modules(&mut self) -> Result<(), ModuleError> {
        const MAX_AUTO_LOAD_MODULES: usize = 32;
        if self.enabled_modules.len() > MAX_AUTO_LOAD_MODULES {
            return Err(ModuleError::ResourceLimitExceeded(format!(
                "enabled_modules has {} entries; auto-load refuses more than {MAX_AUTO_LOAD_MODULES}",
                self.enabled_modules.len()
            )));
        }
        info!("Auto-discovering and loading modules");

        let discovery = ModuleDiscovery::new(&self.spawner.modules_dir);
        let mut discovered_modules = discovery.discover_modules()?;

        if !self.enabled_modules.is_empty() && !self.disabled_modules.is_empty() {
            let disabled_set: std::collections::HashSet<&str> =
                self.disabled_modules.iter().map(|s| s.as_str()).collect();
            for name in self.enabled_modules.keys() {
                if disabled_set.contains(name.as_str()) {
                    warn!(
                        "Module '{}' is both enabled_modules and disabled_modules; disabled wins (will not load or bootstrap)",
                        name
                    );
                }
            }
        }

        // Opt-out: drop explicitly disabled modules before allowlist and bootstrap.
        discovered_modules = self.filter_discovered_for_auto_load(discovered_modules);

        // Bootstrap: for any enabled_modules not yet installed locally (or wrong version), attempt a direct
        // HTTP download from the registry before giving up.
        #[cfg(feature = "governance")]
        if !self.enabled_modules.is_empty() {
            let already_found: std::collections::HashSet<&str> = discovered_modules
                .iter()
                .map(|m| m.manifest.name.as_str())
                .collect();
            let missing: Vec<(String, String)> = self
                .enabled_modules
                .iter()
                .filter(|(name, _)| {
                    !already_found.contains(name.as_str()) && !self.disabled_modules.contains(name)
                })
                .map(|(name, constraint)| (name.clone(), constraint.clone()))
                .collect();

            if !missing.is_empty() {
                if let Some(ref url) = self.registry_url.clone() {
                    info!(
                        "Bootstrap: {} enabled module(s) not installed locally, fetching from registry: {:?}",
                        missing.len(),
                        missing
                            .iter()
                            .map(|(n, c)| format!("{n} ({c})"))
                            .collect::<Vec<_>>()
                    );
                    for (name, constraint) in &missing {
                        match self.bootstrap_download_module(name, url, constraint).await {
                            Ok(()) => {
                                let discovery = ModuleDiscovery::new(&self.spawner.modules_dir);
                                match discovery.discover_module(name) {
                                    Ok(dm) => {
                                        info!("Bootstrap: installed and discovered '{}'", name);
                                        discovered_modules.push(dm);
                                    }
                                    Err(e) => warn!(
                                        "Bootstrap: installed '{}' but re-discovery failed: {}",
                                        name, e
                                    ),
                                }
                            }
                            Err(e) => warn!("Bootstrap: could not install '{}': {}", name, e),
                        }
                    }
                } else {
                    info!(
                        "Bootstrap: {} enabled module(s) not installed and no registry_url configured: {:?}",
                        missing.len(),
                        missing.iter().map(|(n, _)| n.as_str()).collect::<Vec<_>>()
                    );
                }
            }
        }

        if discovered_modules.is_empty() {
            info!(
                "No modules to load (0 discovered or all filtered by disabled_modules / enabled_modules)"
            );
            return Ok(());
        }

        // Try to resolve dependencies - if missing and we have a registry, fetch them
        let resolution = match ModuleDependencies::resolve(&discovered_modules) {
            Ok(res) => res,
            Err(ModuleError::DependencyMissing(msg)) => {
                // Try fetching missing dependencies from registry
                if let Some(ref registry) = self.module_registry {
                    // Parse missing dependencies from error message
                    // Format: "Missing dependencies: [\"dep1\", \"dep2\"]"
                    let missing: Vec<String> = if let Some(start) = msg.find('[') {
                        let deps_str = &msg[start + 1..msg.len() - 1];
                        deps_str
                            .split(',')
                            .map(|s| s.trim().trim_matches('"').to_string())
                            .filter(|s| !s.is_empty())
                            .collect()
                    } else {
                        Vec::new()
                    };

                    // Fetch each missing dependency
                    for dep_name in &missing {
                        info!(
                            "Attempting to fetch missing dependency {} from registry",
                            dep_name
                        );
                        if let Ok(entry) = registry.fetch_module(dep_name).await {
                            // Install fetched module
                            if let Ok(installed) = self.install_module_from_registry(entry).await {
                                discovered_modules.push(installed);
                                info!(
                                    "Successfully fetched and installed module {} from registry",
                                    dep_name
                                );
                            }
                        }
                    }

                    // Re-resolve dependencies after fetching
                    ModuleDependencies::resolve(&discovered_modules)?
                } else {
                    return Err(ModuleError::DependencyMissing(msg));
                }
            }
            Err(e) => return Err(e),
        };

        // Publish ModuleDiscovered for each module (for registry/marketplace sync)
        {
            use crate::module::ipc::protocol::EventPayload;
            use crate::module::traits::EventType;
            for module in &discovered_modules {
                let payload = EventPayload::ModuleDiscovered {
                    module_name: module.manifest.name.clone(),
                    version: module.manifest.version.clone(),
                    source: "filesystem".to_string(),
                };
                if let Err(e) = self
                    .event_manager
                    .publish_event(EventType::ModuleDiscovered, payload)
                    .await
                {
                    warn!(
                        "Failed to publish ModuleDiscovered for {}: {}",
                        module.manifest.name, e
                    );
                }
            }
        }

        // Load module configurations (merge module config.toml with node [modules.<name>] overrides)
        let mut module_configs = HashMap::new();
        for module in &discovered_modules {
            let config_path = module.directory.join("config.toml");
            let base = ModuleLoader::load_module_config(&module.manifest.name, config_path)
                .unwrap_or_default();
            let merged = self.merge_module_config(&module.manifest.name, base);
            module_configs.insert(module.manifest.name.clone(), merged);
        }

        // Load modules in dependency order
        ModuleLoader::load_modules_in_order(
            self,
            &discovered_modules,
            &resolution.load_order,
            &module_configs,
        )
        .await?;

        info!("Auto-loaded {} modules", discovered_modules.len());
        Ok(())
    }

    /// Fetch and install a module from the registry
    pub async fn fetch_module_from_registry(
        &mut self,
        module_name: &str,
    ) -> Result<(), ModuleError> {
        let registry = self.module_registry.as_ref().ok_or_else(|| {
            ModuleError::OperationError(
                crate::module::traits::module_error_msg::MODULE_REGISTRY_NOT_AVAILABLE.to_string(),
            )
        })?;

        info!("Fetching module {} from registry", module_name);
        let entry = registry.fetch_module(module_name).await?;

        // Install the module
        self.install_module_from_registry(entry).await?;

        Ok(())
    }

    /// Install a module entry from the registry to the modules directory
    async fn install_module_from_registry(
        &self,
        entry: crate::module::registry::client::ModuleEntry,
    ) -> Result<crate::module::registry::discovery::DiscoveredModule, ModuleError> {
        use std::fs;
        use std::io::Write;

        // Create module directory
        let module_dir = self.modules_dir.join(&entry.name);
        fs::create_dir_all(&module_dir)
            .map_err(|e| ModuleError::op_err("Failed to create module directory", e))?;

        // Write manifest
        let manifest_path = module_dir.join("module.toml");
        let manifest_toml = toml::to_string_pretty(&entry.manifest)
            .map_err(|e| ModuleError::op_err("Failed to serialize manifest", e))?;
        fs::write(&manifest_path, manifest_toml)
            .map_err(|e| ModuleError::op_err("Failed to write manifest", e))?;

        // Write binary
        let binary_path = module_dir.join(&entry.name);
        if let Some(binary_data) = entry.binary {
            let mut file = fs::File::create(&binary_path)
                .map_err(|e| ModuleError::op_err("Failed to create binary file", e))?;
            file.write_all(&binary_data)
                .map_err(|e| ModuleError::op_err("Failed to write binary", e))?;
        } else {
            // Binary not included, fetch separately if needed
            warn!(
                "Module {} binary not included in registry entry",
                entry.name
            );
        }

        // Make binary executable (Unix)
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut perms = fs::metadata(&binary_path)
                .map_err(|e| ModuleError::op_err("Failed to get file metadata", e))?
                .permissions();
            perms.set_mode(0o755);
            fs::set_permissions(&binary_path, perms)
                .map_err(|e| ModuleError::op_err("Failed to set executable permissions", e))?;
        }

        // Create DiscoveredModule
        let discovery = ModuleDiscovery::new(&self.modules_dir);
        discovery.discover_module(&entry.name)
    }

    /// Download and install a module binary during first-boot bootstrap.
    ///
    /// # Flow (GitHub Releases — `sha256sums.txt` is source of truth)
    ///
    /// 1. Fetch `registry_url` (`modules.json`) → `repo` (`owner/name`); manifest URL is
    ///    `module_toml_url` if set, else `https://raw.githubusercontent.com/{repo}/{ref}/module.toml`
    ///    (`ref` defaults to `main`, overridable with `manifest_ref`).
    /// 2. Fetch `module.toml` → `name`, `version` (no `[downloads]` required).
    /// 3. Tag `v{version}` → download `sha256sums.txt`, then the platform binary
    ///    `{name}-{platform}` (`.exe` on Windows) from the same release.
    /// 4. Verify SHA-256 from the checksums file, install under `modules_dir/<name>/`,
    ///    write `module.toml`.
    #[cfg(feature = "governance")]
    async fn bootstrap_download_module(
        &self,
        name: &str,
        registry_url: &str,
        version_constraint: &str,
    ) -> Result<(), ModuleError> {
        use crate::module::registry::manifest::ModuleManifest;
        use serde::Deserialize;

        // Discovery index: required `name` + `repo` (`owner/name`). Manifest location follows a
        // Cargo.toml-style convention: repo-root `module.toml` on `main`, unless overridden.
        #[derive(Deserialize)]
        struct RegistryEntry {
            name: String,
            repo: String,
            #[serde(default)]
            module_toml_url: Option<String>,
            /// Branch or tag for the default raw GitHub manifest URL (default `main`).
            #[serde(default)]
            manifest_ref: Option<String>,
        }

        let client = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(60))
            .user_agent(concat!("blvm-node/", env!("CARGO_PKG_VERSION")))
            .build()
            .map_err(|e| ModuleError::op_err("Failed to build HTTP client", e))?;

        let fetch = |url: String| {
            let c = client.clone();
            async move {
                let resp = c
                    .get(&url)
                    .send()
                    .await
                    .map_err(|e| ModuleError::op_err("HTTP request failed", e))?;
                if !resp.status().is_success() {
                    return Err(ModuleError::OperationError(format!(
                        "GET {url} returned HTTP {}",
                        resp.status()
                    )));
                }
                resp.bytes()
                    .await
                    .map(|b| b.to_vec())
                    .map_err(|e| ModuleError::op_err("Reading response body", e))
            }
        };

        // ── Step 1: fetch the discovery index ────────────────────────────────
        let index_bytes = fetch(registry_url.to_string()).await?;
        let entries: Vec<RegistryEntry> = serde_json::from_slice(&index_bytes)
            .map_err(|e| ModuleError::op_err("Registry JSON parse failed", e))?;

        let entry = entries.into_iter().find(|e| e.name == name);
        let (github_repo, module_toml_url, manifest_ref_override) = if let Some(entry) = entry {
            (entry.repo, entry.module_toml_url, entry.manifest_ref)
        } else if let Some(repo) =
            crate::module::github_release_install::builtin_default_module_repo(name)
        {
            info!(
                "Bootstrap: '{name}' is a default module and is not in {registry_url}; using {repo}"
            );
            (repo.to_string(), None, None)
        } else {
            return Err(ModuleError::OperationError(format!(
                "Module '{name}' not found in registry at {registry_url}"
            )));
        };

        crate::module::github_release_install::validate_github_repo(&github_repo)?;

        use crate::module::github_release_install::{
            DEFAULT_MODULE_MANIFEST_REF, default_module_toml_raw_url, release_tag,
            resolve_release_version_for_constraint,
        };

        let constraint = version_constraint.trim();
        let pinned_version = if constraint.is_empty() || constraint == "*" {
            None
        } else {
            Some(resolve_release_version_for_constraint(&client, &github_repo, constraint).await?)
        };

        let manifest_ref = if let Some(ref version) = pinned_version {
            release_tag(version)
        } else {
            manifest_ref_override
                .as_deref()
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .unwrap_or(DEFAULT_MODULE_MANIFEST_REF)
                .to_string()
        };

        let module_toml_url = module_toml_url
            .as_deref()
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(|s| s.to_string())
            .unwrap_or_else(|| default_module_toml_raw_url(&github_repo, &manifest_ref));

        // ── Step 2: fetch module.toml from the module's own repo ─────────────
        info!(
            "Bootstrap: fetching manifest for '{}' from {} (constraint: '{}')",
            name, module_toml_url, version_constraint
        );
        let toml_bytes = fetch(module_toml_url.clone()).await?;
        let toml_str = std::str::from_utf8(&toml_bytes)
            .map_err(|e| ModuleError::op_err("module.toml is not valid UTF-8", e))?;
        let mut manifest: ModuleManifest = toml::from_str(toml_str)
            .map_err(|e| ModuleError::op_err("Failed to parse module.toml", e))?;

        if let Some(ref version) = pinned_version {
            let installed = crate::module::github_release_install::manifest_version_for_release(
                &manifest.version,
                Some(version),
            );
            if installed != manifest.version {
                // The tag is cut before module.toml is bumped. The release
                // assets and sha256sums.txt are still this version.
                warn!(
                    "Bootstrap: release {version} of '{name}' has module.toml version {}; installing {version}",
                    manifest.version
                );
                manifest.version = installed;
            }
            if !crate::module::version_constraint::matches_version_constraint(
                &manifest.version,
                constraint,
            ) {
                return Err(ModuleError::OperationError(format!(
                    "Bootstrap: resolved version {} does not satisfy constraint '{}'",
                    manifest.version, version_constraint
                )));
            }
        }

        // ── Step 3: platform + release layout (GitHub Releases + sha256sums.txt) ─
        use crate::module::github_release_install::{
            artifact_name, fetch_release_checksums_text, host_platform_key, release_download_url,
            sha256_from_checksums,
        };

        let platform = host_platform_key()?;

        // On Windows the installed binary must carry the .exe extension so the
        // OS can execute it via Command::new.
        #[cfg(target_os = "windows")]
        let binary_filename = format!("{name}.exe");
        #[cfg(not(target_os = "windows"))]
        let binary_filename = name.to_string();

        let tag = release_tag(&manifest.version);
        let artifact = artifact_name(name, platform)?;
        let checksum_text = fetch_release_checksums_text(&client, &github_repo, &tag).await?;
        let expected_sha = sha256_from_checksums(&checksum_text, &artifact)?;

        let binary_url = release_download_url(&github_repo, &tag, &artifact);
        info!(
            "Bootstrap: downloading '{}' {} for {} from GitHub release {} ({})",
            name, manifest.version, platform, tag, binary_url
        );
        let binary_bytes = fetch(binary_url.clone()).await?;

        use sha2::Digest;
        let actual = hex::encode(sha2::Sha256::digest(&binary_bytes));
        if actual != expected_sha {
            return Err(ModuleError::OperationError(format!(
                "SHA256 mismatch for '{name}' ({platform}) release artifact {artifact}: \
                 expected {expected_sha} got {actual}"
            )));
        }
        debug!("Bootstrap: SHA256 verified for '{}' via {}", name, tag);

        // ── Step 5: write binary + module.toml to modules_dir ────────────────
        let module_dir = self.modules_dir.join(name);
        tokio::fs::create_dir_all(&module_dir)
            .await
            .map_err(|e| ModuleError::op_err("Failed to create module dir", e))?;

        let binary_path = module_dir.join(&binary_filename);
        tokio::fs::write(&binary_path, &binary_bytes)
            .await
            .map_err(|e| ModuleError::op_err("Failed to write binary", e))?;

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut perms = tokio::fs::metadata(&binary_path)
                .await
                .map_err(|e| ModuleError::op_err("Failed to read binary metadata", e))?
                .permissions();
            perms.set_mode(perms.mode() | 0o111);
            tokio::fs::set_permissions(&binary_path, perms)
                .await
                .map_err(|e| ModuleError::op_err("Failed to set executable bit", e))?;
        }

        // Merge `[binary]` (hash + size) into the manifest so load-time checks can use it offline.
        let mut root: toml::Table = toml::from_str(toml_str)
            .map_err(|e| ModuleError::op_err("Failed to parse module.toml for install merge", e))?;
        root.insert(
            "version".into(),
            toml::Value::String(manifest.version.clone()),
        );
        let size_i64 = i64::try_from(binary_bytes.len()).map_err(|_| {
            ModuleError::OperationError("Module binary size does not fit i64".to_string())
        })?;
        let mut binary_tbl = toml::Table::new();
        binary_tbl.insert("hash".into(), toml::Value::String(expected_sha.clone()));
        binary_tbl.insert("size".into(), toml::Value::Integer(size_i64));
        root.insert("binary".into(), toml::Value::Table(binary_tbl));
        let toml_out = toml::to_string(&toml::Value::Table(root))
            .map_err(|e| ModuleError::op_err("Failed to serialize module.toml with [binary]", e))?;
        tokio::fs::write(module_dir.join("module.toml"), toml_out.as_bytes())
            .await
            .map_err(|e| ModuleError::op_err("Failed to write module.toml", e))?;

        info!(
            "Bootstrap: installed '{}' {} to {:?}",
            name, manifest.version, module_dir
        );
        Ok(())
    }

    /// Get event manager for publishing events
    pub fn event_manager(&self) -> &Arc<EventManager> {
        &self.event_manager
    }

    /// Parse permissions from module metadata
    fn parse_permissions_from_metadata(metadata: &ModuleMetadata) -> PermissionSet {
        use crate::module::security::permissions::PermissionSet;

        let mut permissions = PermissionSet::new();

        // Parse permissions from metadata.capabilities (Vec<String>)
        // Note: In module.toml, these are declared as "capabilities" but represent permissions
        use crate::module::security::permissions::parse_permission_string;

        for perm_str in &metadata.capabilities {
            if let Some(permission) = parse_permission_string(perm_str) {
                permissions.add(permission);
            } else {
                warn!("Unknown permission string: {}", perm_str);
            }
        }

        permissions
    }

    /// Stop all modules and shutdown manager
    pub async fn shutdown(&mut self) -> Result<(), ModuleError> {
        info!("Shutting down module manager");

        // Unload all modules
        let module_names: Vec<String> = {
            let modules = self.modules.lock().await;
            modules.keys().cloned().collect()
        };

        for module_name in module_names {
            if let Err(e) = self.unload_module(&module_name).await {
                warn!("Error unloading module {}: {}", module_name, e);
            }
        }

        // Stop IPC server (Unix only)
        #[cfg(unix)]
        if let Some(handle) = self.ipc_server_handle.take() {
            handle.abort();
        }

        info!("Module manager shut down");
        Ok(())
    }

    /// Check if a version satisfies a version requirement
    ///
    /// Supports basic semver constraints:
    /// - Exact: "1.0.0"
    /// - Greater than or equal: ">=1.0.0"
    /// - Less than or equal: "<=1.0.0"
    /// - Range: ">=1.0.0,<2.0.0"
    /// - Wildcard: "1.x" or "1.0.x"
    fn check_version_constraint(version: &str, requirement: &str) -> bool {
        // Simple implementation - for production, consider using semver crate
        // For now, handle common cases

        // Exact match
        if version == requirement {
            return true;
        }

        // Remove whitespace
        let req = requirement.trim();

        // Handle >= constraint
        if let Some(required_version) = req.strip_prefix(">=") {
            return Self::compare_versions(version, required_version.trim()) >= 0;
        }

        // Handle <= constraint
        if let Some(required_version) = req.strip_prefix("<=") {
            return Self::compare_versions(version, required_version.trim()) <= 0;
        }

        // Handle > constraint
        if let Some(required_version) = req.strip_prefix(">") {
            return Self::compare_versions(version, required_version.trim()) > 0;
        }

        // Handle < constraint
        if let Some(required_version) = req.strip_prefix("<") {
            return Self::compare_versions(version, required_version.trim()) < 0;
        }

        // Handle range (comma-separated)
        if req.contains(',') {
            let parts: Vec<&str> = req.split(',').collect();
            return parts
                .iter()
                .all(|part| Self::check_version_constraint(version, part.trim()));
        }

        // Handle wildcard (x or *)
        if req.contains('x') || req.contains('*') {
            // Semver wildcard: prefix match on dot-separated components.
            let version_parts: Vec<&str> = version.split('.').collect();
            let req_parts: Vec<&str> = req.split('.').collect();

            if version_parts.len() != req_parts.len() {
                return false;
            }

            for (v_part, r_part) in version_parts.iter().zip(req_parts.iter()) {
                if *r_part != "x" && *r_part != "*" && v_part != r_part {
                    return false;
                }
            }
            return true;
        }

        // Default: exact match (already checked above, so false)
        false
    }

    #[cfg(test)]
    /// Expose merge_module_config for testing database_backend inheritance
    pub(crate) fn test_merge_module_config(
        &self,
        module_name: &str,
        base: HashMap<String, String>,
    ) -> HashMap<String, String> {
        self.merge_module_config(module_name, base)
    }

    /// Compare two version strings
    /// Returns: -1 if v1 < v2, 0 if v1 == v2, 1 if v1 > v2
    fn compare_versions(v1: &str, v2: &str) -> i32 {
        let parts1: Vec<u32> = v1.split('.').filter_map(|s| s.parse().ok()).collect();
        let parts2: Vec<u32> = v2.split('.').filter_map(|s| s.parse().ok()).collect();

        let max_len = parts1.len().max(parts2.len());

        for i in 0..max_len {
            let p1 = parts1.get(i).copied().unwrap_or(0);
            let p2 = parts2.get(i).copied().unwrap_or(0);

            if p1 < p2 {
                return -1;
            } else if p1 > p2 {
                return 1;
            }
        }

        0
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::module::traits::ModuleMetadata;
    use std::collections::HashMap;

    #[test]
    fn test_database_backend_inheritance() {
        let temp = std::env::temp_dir().join("blvm_module_test");
        std::fs::create_dir_all(&temp).unwrap();
        let manager = ModuleManager::new(&temp, &temp, &temp);
        manager.set_default_database_backend("sled".to_string());

        let merged = manager.test_merge_module_config("my-module", HashMap::new());
        assert_eq!(
            merged.get("database_backend"),
            Some(&"sled".to_string()),
            "module should inherit effective subprocess KV backend when not set"
        );
    }

    #[test]
    fn test_database_backend_no_override_when_module_sets() {
        let temp = std::env::temp_dir().join("blvm_module_test2");
        std::fs::create_dir_all(&temp).unwrap();
        let manager = ModuleManager::new(&temp, &temp, &temp);
        manager.set_default_database_backend("rocksdb".to_string());

        let mut base = HashMap::new();
        base.insert("database_backend".to_string(), "redb".to_string());
        let merged = manager.test_merge_module_config("my-module", base);
        assert_eq!(
            merged.get("database_backend"),
            Some(&"redb".to_string()),
            "module's database_backend should not be overridden by node default"
        );
    }

    /// When wasm-modules feature is disabled, loading a .wasm file must return a clear error.
    #[cfg(not(feature = "wasm-modules"))]
    #[tokio::test]
    async fn test_wasm_load_requires_feature_when_disabled() {
        let temp = tempfile::tempdir().expect("tempdir");
        let temp_path = temp.path();
        let wasm_path = temp_path.join("fake.wasm");
        std::fs::write(&wasm_path, b"").unwrap();

        let mut manager = ModuleManager::new(temp_path, temp_path, temp_path);
        let metadata = ModuleMetadata {
            name: "fake".to_string(),
            version: "0.1.0".to_string(),
            description: String::new(),
            author: String::new(),
            capabilities: vec![],
            rpc_overrides: vec![],
            dependencies: HashMap::new(),
            optional_dependencies: HashMap::new(),
            entry_point: "fake".to_string(),
        };

        let err = manager
            .load_module("fake", &wasm_path, metadata, HashMap::new())
            .await
            .unwrap_err();

        let msg = err.to_string();
        assert!(
            msg.contains("wasm-modules"),
            "expected error to mention wasm-modules, got: {msg}"
        );
    }

    fn fake_discovered(
        name: &str,
        version: &str,
    ) -> crate::module::registry::discovery::DiscoveredModule {
        use crate::module::registry::discovery::DiscoveredModule;
        use std::path::PathBuf;

        let manifest: crate::module::registry::manifest::ModuleManifest = toml::from_str(&format!(
            r#"name = "{name}"
version = "{version}"
entry_point = "{name}"
"#
        ))
        .unwrap();
        DiscoveredModule {
            directory: PathBuf::from("/unused"),
            manifest,
            binary_path: PathBuf::from("/unused/bin"),
        }
    }

    #[test]
    fn filter_discovered_allowlist_and_version_constraint() {
        let temp = tempfile::tempdir().unwrap();
        let mut manager = ModuleManager::new(temp.path(), temp.path(), temp.path());
        manager.set_enabled_modules(
            [
                ("alpha".into(), "0.1.*".into()),
                ("beta".into(), "2.0.0".into()),
            ]
            .into_iter()
            .collect(),
        );

        let discovered = vec![
            fake_discovered("alpha", "0.1.3"),
            fake_discovered("alpha", "0.2.0"),
            fake_discovered("beta", "2.0.0"),
            fake_discovered("gamma", "1.0.0"),
        ];

        let filtered = manager.filter_discovered_for_auto_load(discovered);
        let names: Vec<_> = filtered.iter().map(|m| m.manifest.name.as_str()).collect();
        assert_eq!(names, vec!["alpha", "beta"]);
        assert_eq!(filtered[0].manifest.version, "0.1.3");
        assert_eq!(filtered[1].manifest.version, "2.0.0");
    }

    #[test]
    fn filter_discovered_disabled_modules_win() {
        let temp = tempfile::tempdir().unwrap();
        let mut manager = ModuleManager::new(temp.path(), temp.path(), temp.path());
        manager.set_enabled_modules([("alpha".into(), "*".into())].into_iter().collect());
        manager.set_disabled_modules(vec!["alpha".into()]);

        let filtered =
            manager.filter_discovered_for_auto_load(vec![fake_discovered("alpha", "0.1.0")]);
        assert!(filtered.is_empty());
    }

    #[tokio::test]
    async fn unload_clears_the_module_metric_series() {
        use crate::module::metrics::manager::{Metric, MetricValue};

        let temp = tempfile::tempdir().unwrap();
        let mut manager = ModuleManager::new(temp.path(), temp.path(), temp.path());
        let metrics = Arc::new(crate::module::metrics::MetricsManager::new());
        manager.set_metrics_manager(Arc::clone(&metrics));
        let module_id = "fees_0123456789abcdef0123456789abcdef".to_string();
        metrics
            .report_metric(
                module_id.clone(),
                Metric {
                    name: "depth".into(),
                    value: MetricValue::Gauge(1.0),
                    labels: HashMap::new(),
                    timestamp: 1,
                },
            )
            .await;
        {
            let mut modules = manager.modules.lock().await;
            modules.insert(
                "fees".into(),
                ManagedModule {
                    metadata: ModuleMetadata {
                        name: "fees".into(),
                        version: "0.1.0".into(),
                        description: String::new(),
                        author: String::new(),
                        capabilities: Vec::new(),
                        rpc_overrides: Vec::new(),
                        dependencies: HashMap::new(),
                        optional_dependencies: HashMap::new(),
                        entry_point: "fees".into(),
                    },
                    process: None,
                    state: ModuleState::Stopped,
                    monitor_handle: None,
                    process_id: None,
                    module_id: module_id.clone(),
                    #[cfg(feature = "wasm-modules")]
                    wasm_instance: None,
                },
            );
        }
        manager.unload_module("fees").await.unwrap();
        assert!(metrics.get_module_metrics(&module_id).await.is_empty());
    }

    #[test]
    fn filter_discovered_empty_enabled_passes_all() {
        let temp = tempfile::tempdir().unwrap();
        let manager = ModuleManager::new(temp.path(), temp.path(), temp.path());
        let discovered = vec![fake_discovered("alpha", "9.9.9")];
        let filtered = manager.filter_discovered_for_auto_load(discovered);
        assert_eq!(filtered.len(), 1);
    }
}