//! 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", "tidesdb".
    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>,
    /// 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>>,
}

/// 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>,
    /// 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
    }
}

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,
            node_p2p_listen: std::sync::RwLock::new(None),
        }
    }

    /// 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);
    }

    /// 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;
    }

    /// 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,
        ))));
        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));
            #[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();
            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();
        #[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 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();
                            }
                            // Update state to Error
                            managed.state = ModuleState::Error(error.to_string());
                        }
                    }

                    // Clean up: loaded_modules, CLI spec, API hub (same as explicit unload)
                    event_manager.remove_loaded_module(&module_name).await;
                    #[cfg(unix)]
                    if let Some(ref ipc) = ipc_server {
                        ipc.lock()
                            .await
                            .unregister_cli_spec_by_name(&module_name)
                            .await;
                    }
                    if let Some(ref hub) = api_hub {
                        let mut h = hub.lock().await;
                        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;
                            #[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,
            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![],
                        });
                let wasm_instance = Some(Arc::new(tokio::sync::Mutex::new(instance)));

                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: None,
                    state: ModuleState::Running,
                    monitor_handle: None,
                    process_id: None,
                    wasm_instance,
                };
                modules.insert(module_name.to_string(), managed);
                drop(modules);
                if let Some(ref ipc) = self.ipc_server {
                    ipc.lock()
                        .await
                        .register_cli_spec(module_name.to_string(), cli_spec)
                        .await;
                }
                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 monitor_handle = tokio::spawn(async move {
            if let Err(e) = monitor
                .monitor_module_shared(module_name_clone.clone(), shared_process_for_monitor)
                .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,
            #[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;
        let module_id = modules
            .get(module_name)
            .map(|_m| format!("{}_{}", module_name, new_request_id())) // We don't store module_id, so generate one
            .unwrap_or_else(|| module_name.to_string());

        if let Some(mut managed) = modules.remove(module_name) {
            // 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);
            {
                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
            if let Some(handle) = managed.monitor_handle.take() {
                handle.abort();
            }

            // 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 {
                // Kill by PID if we don't have process reference
                use tokio::process::Command;
                #[cfg(unix)]
                {
                    let _ = Command::new("kill")
                        .arg("-9")
                        .arg(pid.to_string())
                        .output()
                        .await;
                }
            }

            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 {
                ipc.lock()
                    .await
                    .unregister_cli_spec_by_name(module_name)
                    .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;
                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(())
    }

    /// Validate optional dependencies (soft dependencies)
    /// Returns list of missing optional dependencies (non-fatal)
    pub async fn validate_optional_dependencies(&self, module_name: &str) -> Vec<String> {
        let modules = self.modules.lock().await;
        let module = match modules.get(module_name) {
            Some(m) => m,
            None => return Vec::new(),
        };

        let mut missing = Vec::new();
        for dep_name in module.metadata.optional_dependencies.keys() {
            if !modules.contains_key(dep_name) {
                missing.push(dep_name.clone());
            }
        }

        missing
    }

    /// 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
    }

    /// Check if a module can be safely unloaded (no dependents)
    pub async fn can_unload_module(&self, module_name: &str) -> Result<bool, ModuleError> {
        let dependents = self.get_dependent_modules(module_name).await;
        Ok(dependents.is_empty())
    }

    /// 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> {
        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)
            .ok_or_else(|| {
                ModuleError::OperationError(format!(
                    "Module '{name}' not found in registry at {registry_url}"
                ))
            })?;

        let github_repo = entry.repo;

        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 {
            entry
                .manifest_ref
                .as_deref()
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .unwrap_or(DEFAULT_MODULE_MANIFEST_REF)
                .to_string()
        };

        let module_toml_url = entry
            .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 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 {
            if manifest.version != *version {
                return Err(ModuleError::OperationError(format!(
                    "Bootstrap: module.toml version {} does not match resolved release {} for constraint '{}'",
                    manifest.version, version, version_constraint
                )));
            }
            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))?;
        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());
    }

    #[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);
    }
}
