""" Dynamic Plugin Loader with Lifecycle Management This module provides dynamic loading of plugins from ./plugins/*/ directories with: - Safe plugin module loading and unloading - Dependency resolution and validation - Plugin class discovery and instantiation - Error isolation and recovery - Hot-reload capability for development - Module cleanup and garbage collection - Security validation and sandboxing - Plugin compatibility checking Features: - Discovers plugin directories automatically - Loads plugin modules dynamically using importlib - Instantiates plugin classes that extend TrixyPlugin - Manages plugin lifecycle (load, initialize, enable, disable, unload) - Provides hot-reload for development - Isolates plugin errors from system - Validates plugin compatibility and dependencies - Cleans up resources on unload """ import os import sys import importlib import importlib.util import threading import time import traceback import inspect from typing import Dict, List, Optional, Any, Type, Set, Tuple from pathlib import Path from dataclasses import dataclass from enum import Enum import json # Import plugin base class from .trixy_plugin import TrixyPlugin, PluginState, PluginError from .config_handler import PluginConfigHandler def pprint(message: str) -> None: """Plugin loader logging function.""" print(f"[LOADER] {message}") class LoadResult(Enum): """Plugin load operation results.""" SUCCESS = "success" FAILED = "failed" SKIPPED = "skipped" NOT_FOUND = "not_found" INVALID = "invalid" DEPENDENCY_ERROR = "dependency_error" COMPATIBILITY_ERROR = "compatibility_error" @dataclass class PluginLoadResult: """Result of a plugin load operation.""" plugin_name: str result: LoadResult plugin_instance: Optional[TrixyPlugin] = None error_message: Optional[str] = None load_time: float = 0.0 warnings: List[str] = None def __post_init__(self): if self.warnings is None: self.warnings = [] @property def success(self) -> bool: """Check if load was successful.""" return self.result == LoadResult.SUCCESS @property def failed(self) -> bool: """Check if load failed.""" return self.result == LoadResult.FAILED class PluginLoaderError(Exception): """Base exception for plugin loader errors.""" pass class PluginDiscoveryError(PluginLoaderError): """Raised when plugin discovery fails.""" pass class PluginModuleError(PluginLoaderError): """Raised when plugin module loading fails.""" pass class PluginDependencyError(PluginLoaderError): """Raised when plugin dependencies cannot be resolved.""" pass class PluginLoader: """ Dynamic plugin loader with advanced lifecycle management. This class handles: - Plugin discovery in ./plugins/*/ directories - Dynamic module loading and unloading - Plugin class instantiation - Dependency resolution - Error isolation and recovery - Hot-reload for development - Resource cleanup Plugin Directory Structure: ./plugins/plugin_name/ ├── main.py # Contains plugin class extending TrixyPlugin ├── config.json # Plugin configuration └── config_view.py # Optional: Custom TUI configuration Usage: loader = PluginLoader( plugins_directory=Path("./plugins"), application=application_container ) # Discover plugins discovered = loader.discover_plugins() # Load a specific plugin result = loader.load_plugin("my_plugin") # Hot reload during development result = loader.reload_plugin("my_plugin") # Unload plugin loader.unload_plugin("my_plugin") """ def __init__( self, plugins_directory: Path, application: Any, auto_reload: bool = False, validate_compatibility: bool = True, enable_sandboxing: bool = False ): """ Initialize the plugin loader. Args: plugins_directory: Directory containing plugin subdirectories application: Application container instance auto_reload: Enable automatic reloading when files change validate_compatibility: Validate plugin compatibility enable_sandboxing: Enable plugin sandboxing (future feature) """ self.plugins_directory = Path(plugins_directory) self.application = application self.auto_reload = auto_reload self.validate_compatibility = validate_compatibility self.enable_sandboxing = enable_sandboxing # Loaded plugins tracking self._loaded_plugins: Dict[str, TrixyPlugin] = {} self._loaded_modules: Dict[str, Any] = {} self._plugin_paths: Dict[str, Path] = {} self._load_lock = threading.RLock() # Dependency tracking self._plugin_dependencies: Dict[str, Set[str]] = {} self._dependency_graph: Dict[str, Set[str]] = {} # File modification tracking for hot-reload self._file_timestamps: Dict[str, float] = {} self._watch_thread: Optional[threading.Thread] = None self._watch_shutdown = threading.Event() # Statistics self._load_stats = { 'total_loads': 0, 'successful_loads': 0, 'failed_loads': 0, 'reload_count': 0, 'discovery_count': 0 } # Ensure plugins directory exists self.plugins_directory.mkdir(parents=True, exist_ok=True) pprint(f"PluginLoader initialized for directory: {plugins_directory}") # Start file watcher if auto-reload enabled if self.auto_reload: self._start_file_watcher() def discover_plugins(self) -> List[str]: """ Discover all available plugins in the plugins directory. Returns: List of plugin names that were discovered Raises: PluginDiscoveryError: If discovery fails """ try: discovered_plugins = [] if not self.plugins_directory.exists(): pprint(f"Plugins directory does not exist: {self.plugins_directory}") return [] # Scan for plugin directories for item in self.plugins_directory.iterdir(): if item.is_dir() and not item.name.startswith('.'): plugin_name = item.name main_file = item / "main.py" if main_file.exists(): discovered_plugins.append(plugin_name) self._plugin_paths[plugin_name] = item pprint(f"Discovered plugin: {plugin_name}") else: pprint(f"Skipping {plugin_name}: no main.py found") self._load_stats['discovery_count'] = len(discovered_plugins) pprint(f"Discovery complete: found {len(discovered_plugins)} plugins") return discovered_plugins except Exception as e: error_msg = f"Plugin discovery failed: {e}" pprint(error_msg) raise PluginDiscoveryError(error_msg) from e def load_plugin(self, plugin_name: str, force_reload: bool = False) -> PluginLoadResult: """ Load a specific plugin. Args: plugin_name: Name of the plugin to load force_reload: Force reload if already loaded Returns: PluginLoadResult with load status and details """ start_time = time.time() with self._load_lock: try: # Check if already loaded if plugin_name in self._loaded_plugins and not force_reload: return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.SKIPPED, plugin_instance=self._loaded_plugins[plugin_name], warnings=["Plugin already loaded"] ) # Get plugin path if plugin_name not in self._plugin_paths: # Try to discover it discovered = self.discover_plugins() if plugin_name not in discovered: return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.NOT_FOUND, error_message=f"Plugin {plugin_name} not found in {self.plugins_directory}" ) plugin_path = self._plugin_paths[plugin_name] # Unload if already loaded (for reload) if plugin_name in self._loaded_plugins: self._unload_plugin_internal(plugin_name) # Load plugin module module = self._load_plugin_module(plugin_name, plugin_path) if module is None: return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.FAILED, error_message="Failed to load plugin module" ) # Find plugin class plugin_class = self._find_plugin_class(module, plugin_name) if plugin_class is None: return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.INVALID, error_message="No valid TrixyPlugin class found in main.py" ) # Load plugin configuration config_handler = PluginConfigHandler( config_path=plugin_path / "config.json", plugin_name=plugin_name ) initial_config = config_handler.load_config() # Validate dependencies dependencies = initial_config.get('dependencies', []) if dependencies and not self._check_dependencies(plugin_name, dependencies): return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.DEPENDENCY_ERROR, error_message=f"Unmet dependencies: {dependencies}" ) # Validate compatibility if self.validate_compatibility: compat_result = self._check_compatibility(plugin_name, initial_config) if not compat_result[0]: return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.COMPATIBILITY_ERROR, error_message=f"Compatibility check failed: {compat_result[1]}" ) # Instantiate plugin plugin_instance = plugin_class( plugin_name=plugin_name, plugin_path=plugin_path, application=self.application, initial_config=initial_config ) # Store loaded plugin self._loaded_plugins[plugin_name] = plugin_instance self._loaded_modules[plugin_name] = module self._plugin_dependencies[plugin_name] = set(dependencies) # Update file timestamps for hot-reload if self.auto_reload: self._update_file_timestamps(plugin_path) # Update statistics self._load_stats['total_loads'] += 1 self._load_stats['successful_loads'] += 1 load_time = time.time() - start_time pprint(f"Plugin {plugin_name} loaded successfully in {load_time:.3f}s") return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.SUCCESS, plugin_instance=plugin_instance, load_time=load_time ) except Exception as e: self._load_stats['total_loads'] += 1 self._load_stats['failed_loads'] += 1 error_msg = f"Failed to load plugin {plugin_name}: {e}" pprint(error_msg) pprint(traceback.format_exc()) return PluginLoadResult( plugin_name=plugin_name, result=LoadResult.FAILED, error_message=error_msg, load_time=time.time() - start_time ) def load_all_plugins(self) -> Dict[str, PluginLoadResult]: """ Load all discovered plugins. Returns: Dict mapping plugin names to their load results """ discovered_plugins = self.discover_plugins() results = {} # Sort plugins by dependencies (load dependencies first) sorted_plugins = self._sort_plugins_by_dependencies(discovered_plugins) for plugin_name in sorted_plugins: result = self.load_plugin(plugin_name) results[plugin_name] = result if result.success: pprint(f"✓ {plugin_name}") else: pprint(f"✗ {plugin_name}: {result.error_message}") successful = len([r for r in results.values() if r.success]) pprint(f"Loaded {successful}/{len(results)} plugins successfully") return results def unload_plugin(self, plugin_name: str) -> bool: """ Unload a specific plugin. Args: plugin_name: Name of plugin to unload Returns: bool: True if successfully unloaded """ with self._load_lock: return self._unload_plugin_internal(plugin_name) def _unload_plugin_internal(self, plugin_name: str) -> bool: """Internal plugin unloading logic.""" try: if plugin_name in self._loaded_plugins: plugin_instance = self._loaded_plugins[plugin_name] # Call plugin unload lifecycle method plugin_instance.unload_plugin() # Remove from tracking del self._loaded_plugins[plugin_name] if plugin_name in self._loaded_modules: del self._loaded_modules[plugin_name] if plugin_name in self._plugin_dependencies: del self._plugin_dependencies[plugin_name] pprint(f"Plugin {plugin_name} unloaded successfully") return True else: pprint(f"Plugin {plugin_name} not loaded") return False except Exception as e: pprint(f"Error unloading plugin {plugin_name}: {e}") return False def unload_all_plugins(self) -> Dict[str, bool]: """ Unload all loaded plugins. Returns: Dict mapping plugin names to unload success status """ results = {} # Unload in reverse dependency order plugin_names = list(self._loaded_plugins.keys()) sorted_plugins = self._sort_plugins_by_dependencies(plugin_names, reverse=True) for plugin_name in sorted_plugins: results[plugin_name] = self.unload_plugin(plugin_name) return results def reload_plugin(self, plugin_name: str) -> PluginLoadResult: """ Reload a plugin (hot-reload). Args: plugin_name: Name of plugin to reload Returns: PluginLoadResult with reload status """ pprint(f"Reloading plugin {plugin_name}") self._load_stats['reload_count'] += 1 # Force reload return self.load_plugin(plugin_name, force_reload=True) def _load_plugin_module(self, plugin_name: str, plugin_path: Path) -> Optional[Any]: """ Load a plugin module dynamically. Args: plugin_name: Name of the plugin plugin_path: Path to the plugin directory Returns: Loaded module or None if failed """ try: main_file = plugin_path / "main.py" # Create module name module_name = f"trixy_plugin_{plugin_name}" # Load module spec spec = importlib.util.spec_from_file_location(module_name, main_file) if spec is None or spec.loader is None: pprint(f"Failed to create module spec for {plugin_name}") return None # Create module module = importlib.util.module_from_spec(spec) # Add to sys.modules temporarily for imports to work sys.modules[module_name] = module # Execute module spec.loader.exec_module(module) pprint(f"Module {module_name} loaded successfully") return module except Exception as e: pprint(f"Failed to load module for plugin {plugin_name}: {e}") # Clean up sys.modules if we added it module_name = f"trixy_plugin_{plugin_name}" if module_name in sys.modules: del sys.modules[module_name] return None def _find_plugin_class(self, module: Any, plugin_name: str) -> Optional[Type[TrixyPlugin]]: """ Find the TrixyPlugin class in the loaded module. Args: module: Loaded plugin module plugin_name: Name of the plugin Returns: Plugin class or None if not found """ try: # Look for classes that inherit from TrixyPlugin for name, obj in inspect.getmembers(module): if (inspect.isclass(obj) and issubclass(obj, TrixyPlugin) and obj != TrixyPlugin): pprint(f"Found plugin class: {name} in {plugin_name}") return obj pprint(f"No TrixyPlugin subclass found in {plugin_name}") return None except Exception as e: pprint(f"Error finding plugin class in {plugin_name}: {e}") return None def _check_dependencies(self, plugin_name: str, dependencies: List[str]) -> bool: """ Check if plugin dependencies are satisfied. Args: plugin_name: Name of the plugin dependencies: List of required dependency names Returns: bool: True if all dependencies are satisfied """ for dep in dependencies: if dep not in self._loaded_plugins: pprint(f"Plugin {plugin_name} missing dependency: {dep}") return False return True def _check_compatibility(self, plugin_name: str, config: Dict[str, Any]) -> Tuple[bool, str]: """ Check plugin compatibility. Args: plugin_name: Name of the plugin config: Plugin configuration Returns: Tuple of (is_compatible, error_message) """ try: # Check minimum Trixy version min_version = config.get('min_trixy_version', '1.0.0') current_version = getattr(self.application, 'get_version', lambda: '1.0.0')() if self._version_compare(current_version, min_version) < 0: return False, f"Requires Trixy {min_version}, current is {current_version}" # Check Python version if specified min_python = config.get('min_python_version') if min_python: current_python = f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}" if self._version_compare(current_python, min_python) < 0: return False, f"Requires Python {min_python}, current is {current_python}" return True, "" except Exception as e: return False, f"Compatibility check error: {e}" def _version_compare(self, version1: str, version2: str) -> int: """ Compare version strings. Returns: -1 if version1 < version2 0 if version1 == version2 1 if version1 > version2 """ try: v1_parts = [int(x) for x in version1.split('.')] v2_parts = [int(x) for x in version2.split('.')] # Pad shorter version with zeros max_len = max(len(v1_parts), len(v2_parts)) v1_parts += [0] * (max_len - len(v1_parts)) v2_parts += [0] * (max_len - len(v2_parts)) for v1, v2 in zip(v1_parts, v2_parts): if v1 < v2: return -1 elif v1 > v2: return 1 return 0 except Exception: return 0 def _sort_plugins_by_dependencies(self, plugin_names: List[str], reverse: bool = False) -> List[str]: """ Sort plugins by dependency order. Args: plugin_names: List of plugin names to sort reverse: If True, sort in reverse dependency order Returns: Sorted list of plugin names """ # Simple topological sort would go here # For now, just return the original order # This could be enhanced with proper dependency resolution result = plugin_names.copy() if reverse: result.reverse() return result def _start_file_watcher(self) -> None: """Start the file watcher thread for hot-reload.""" if self._watch_thread is None or not self._watch_thread.is_alive(): self._watch_thread = threading.Thread( target=self._file_watcher_loop, name="PluginFileWatcher", daemon=True ) self._watch_thread.start() pprint("File watcher started for hot-reload") def _file_watcher_loop(self) -> None: """File watcher loop for detecting changes.""" while not self._watch_shutdown.wait(2.0): # Check every 2 seconds try: for plugin_name in list(self._loaded_plugins.keys()): if self._check_plugin_files_changed(plugin_name): pprint(f"Files changed for plugin {plugin_name}, reloading...") self.reload_plugin(plugin_name) except Exception as e: pprint(f"Error in file watcher: {e}") def _check_plugin_files_changed(self, plugin_name: str) -> bool: """Check if plugin files have been modified.""" if plugin_name not in self._plugin_paths: return False plugin_path = self._plugin_paths[plugin_name] main_file = plugin_path / "main.py" config_file = plugin_path / "config.json" files_to_check = [main_file] if config_file.exists(): files_to_check.append(config_file) for file_path in files_to_check: if file_path.exists(): current_mtime = file_path.stat().st_mtime stored_mtime = self._file_timestamps.get(str(file_path), 0) if current_mtime > stored_mtime: self._update_file_timestamps(plugin_path) return True return False def _update_file_timestamps(self, plugin_path: Path) -> None: """Update stored file timestamps for a plugin.""" main_file = plugin_path / "main.py" config_file = plugin_path / "config.json" if main_file.exists(): self._file_timestamps[str(main_file)] = main_file.stat().st_mtime if config_file.exists(): self._file_timestamps[str(config_file)] = config_file.stat().st_mtime def shutdown(self) -> None: """Shutdown the plugin loader and clean up resources.""" pprint("Shutting down plugin loader...") # Stop file watcher if self._watch_thread: self._watch_shutdown.set() self._watch_thread.join(timeout=5.0) # Unload all plugins self.unload_all_plugins() # Clean up module references module_names = [f"trixy_plugin_{name}" for name in self._loaded_modules.keys()] for module_name in module_names: if module_name in sys.modules: del sys.modules[module_name] pprint("Plugin loader shutdown complete") # Query and Status Methods def get_loaded_plugins(self) -> Dict[str, TrixyPlugin]: """Get all loaded plugin instances.""" with self._load_lock: return self._loaded_plugins.copy() def is_plugin_loaded(self, plugin_name: str) -> bool: """Check if a plugin is loaded.""" return plugin_name in self._loaded_plugins def get_plugin_info(self, plugin_name: str) -> Optional[Dict[str, Any]]: """Get information about a specific plugin.""" if plugin_name not in self._loaded_plugins: return None plugin = self._loaded_plugins[plugin_name] return { 'name': plugin_name, 'state': plugin.state.value, 'enabled': plugin.enabled, 'path': str(plugin.plugin_path), 'load_time': plugin.load_time, 'init_time': plugin.init_time, 'dependencies': list(self._plugin_dependencies.get(plugin_name, [])), 'metadata': plugin.metadata.__dict__ if plugin.metadata else None, 'health': plugin.check_health() } def get_loader_stats(self) -> Dict[str, Any]: """Get loader statistics.""" return { 'plugins_directory': str(self.plugins_directory), 'auto_reload': self.auto_reload, 'loaded_count': len(self._loaded_plugins), 'discovered_count': len(self._plugin_paths), 'stats': self._load_stats.copy() } def __repr__(self) -> str: """String representation of the plugin loader.""" return f""