plugin_manager.py

# plugin_manager.py (Mythos-Killer v8.0 "मिथोस-सर्वर" — हफ्ता 4)
# यह फ़ाइल हमारे सिस्टम का "प्लग-एंड-प-ले" इंजन है।
# cli.js के mcp add पैटर्न से प्रेरित — बिना कोर कोड बदले कोई भी
# नया AI मॉडल, हैकिंग टूल, या स्किल प्लगइन की तरह जोड़ा जा सकता है।

import sys
import json
import importlib
import threading
from pathlib import Path
from typing import Dict, Any, List, Optional, Callable
from datetime import datetime
from dataclasses import dataclass, field


# हफ्ता 1: EventBus
# FIX (DEAD CODE): Events इम्पोर्ट हटाया — कोड में कहीं उपयोग नहीं था।
try:
    from event_bus import get_event_bus
    EVENT_BUS_AVAILABLE = True
except ImportError:
    EVENT_BUS_AVAILABLE = False


@dataclass
class PluginManifest:
    """एक प्लगइन की परिभाषा (manifest.json से लोड)"""
    name: str
    version: str
    description: str = ""
    author: str = ""
    type: str = "tool"                # tool, model, skill, agent
    entry_point: str = ""             # मुख्य मॉड्यूल/क्लास का नाम
    dependencies: List[str] = field(default_factory=list)
    permissions: List[str] = field(default_factory=list)
    config: Dict[str, Any] = field(default_factory=dict)
    hooks: List[str] = field(default_factory=list)  # pre_task, post_task, on_error
    models: List[Dict[str, Any]] = field(default_factory=list)  # मॉडल प्लगइन के लिए
    tools: List[Dict[str, Any]] = field(default_factory=list)    # टूल प्लगइन के लिए


@dataclass
class LoadedPlugin:
    """लोडेड प्लगइन का रनटाइम रिप्रेजेंटेशन"""
    manifest: PluginManifest
    module: Optional[Any] = None       # Python मॉड्यूल ऑब्जेक्ट
    instance: Optional[Any] = None     # क्लास इंस्टेंस (यदि लागू हो)
    enabled: bool = True
    loaded_at: str = ""
    error: Optional[str] = None


class PluginManager:
    """
    v8.0 प्लगइन मैनेजर — "प्लग-एंड-प्ले" सिस्टम।

    यह बिना कोर कोड बदले नए टूल्स, मॉडल्स और स्किल्स को जोड़ने देता है।
    cli.js के mcp add पैटर्न से प्रेरित।

    उपयोग:
        pm = PluginManager(plugin_dir="/workspace/plugins")
        pm.discover_plugins()
        pm.load_all()
        pm.execute_hook("pre_task", {"task_id": "123"})
    """

    def __init__(
        self,
        plugin_dir: str = "/workspace/plugins",
        event_bus=None,
        auto_discover: bool = True,
    ):
        """
        PluginManager को प्रारंभ करता है।

        पैरामीटर्स:
            plugin_dir: प्लगइन फ़ाइलों की मुख्य डायरेक्टरी
            event_bus: EventBus इंस्टेंस (हफ्ता 1)
            auto_discover: क्या ऑटो-डिस्कवर करना है?
        """
        self.plugin_dir = Path(plugin_dir)
        self.plugin_dir.mkdir(parents=True, exist_ok=True)

        # EventBus
        self.event_bus = event_bus
        if self.event_bus is None and EVENT_BUS_AVAILABLE:
            try:
                self.event_bus = get_event_bus()
            except Exception:
                pass

        # प्लगइन रजिस्ट्री
        self._plugins: Dict[str, LoadedPlugin] = {}
        self._hook_registry: Dict[str, List[Callable]] = {}
        self._model_registry: Dict[str, Dict] = {}
        self._tool_registry: Dict[str, Callable] = {}

        # थ्रेड-सुरक्षा
        self._lock = threading.RLock()

        # स्टैट्स
        self._stats = {
            "total_discovered": 0,
            "total_loaded": 0,
            "total_failed": 0,
            "hooks_registered": 0,
            "started_at": datetime.now().isoformat(),
        }

        # ऑटो-डिस्कवर
        if auto_discover:
            self.discover_plugins()

    # प्लगइन डिस्कवरी

    def discover_plugins(self) -> List[str]:
        """
        प्लगइन डायरेक्टरी से सभी उपलब्ध प्लगइन्स खोजता है।
        हर सब-डायरेक्टरी एक प्लगइन है (जिसमें manifest.json हो)।

        लौटाता है:
            खोजे गए प्लगइन नामों की सूची
        """
        discovered = []

        if not self.plugin_dir.exists():
            print(f"[PluginManager] Plugin directory not found: {self.plugin_dir}")
            return discovered

        for item in self.plugin_dir.iterdir():
            if not item.is_dir():
                continue

            manifest_path = item / "manifest.json"
            if not manifest_path.exists():
                continue

            try:
                with open(manifest_path, 'r', encoding='utf-8') as f:
                    manifest_data = json.load(f)

                manifest = PluginManifest(
                    name=manifest_data.get("name", item.name),
                    version=manifest_data.get("version", "0.0.0"),
                    description=manifest_data.get("description", ""),
                    author=manifest_data.get("author", ""),
                    type=manifest_data.get("type", "tool"),
                    entry_point=manifest_data.get("entry_point", ""),
                    dependencies=manifest_data.get("dependencies", []),
                    permissions=manifest_data.get("permissions", []),
                    config=manifest_data.get("config", {}),
                    hooks=manifest_data.get("hooks", []),
                    models=manifest_data.get("models", []),
                    tools=manifest_data.get("tools", []),
                )

                with self._lock:
                    if manifest.name not in self._plugins:
                        self._plugins[manifest.name] = LoadedPlugin(
                            manifest=manifest,
                            loaded_at="",
                        )
                    discovered.append(manifest.name)
                    self._stats["total_discovered"] += 1

                print(f"[PluginManager] Discovered: {manifest.name} v{manifest.version} ({manifest.type})")

            except json.JSONDecodeError as e:
                print(f"[PluginManager] Invalid manifest.json in {item.name}: {e}")
            except Exception as e:
                print(f"[PluginManager] Error discovering {item.name}: {e}")

        return discovered

    # प्लगइन लोडिंग

    def load_plugin(self, name: str) -> bool:
        """
        एक प्लगइन को लोड करता है।

        पैरामीटर्स:
            name: प्लगइन का नाम

        लौटाता है:
            True अगर सफल, False अगर विफल
        """
        # FIX (MEDIUM): पूरा लोडिंग और इंस्टेंशिएशन प्रोसेस lock के अंदर
        # ताकि एक ही प्लगइन एक साथ दो थ्रेड्स द्वारा लोड न हो सके।
        with self._lock:
            if name not in self._plugins:
                print(f"[PluginManager] Plugin '{name}' not found in registry")
                return False

            plugin = self._plugins[name]
            if plugin.module is not None:
                print(f"[PluginManager] Plugin '{name}' is already loaded")
                return True

            manifest = plugin.manifest
            plugin_path = self.plugin_dir / name

            # Python पथ में प्लगइन डायरेक्टरी जोड़ें
            if str(plugin_path) not in sys.path:
                sys.path.insert(0, str(plugin_path))

            try:
                # एंट्री पॉइंट लोड करें
                if manifest.entry_point:
                    if ":" in manifest.entry_point:
                        module_name, class_name = manifest.entry_point.split(":", 1)
                        module = importlib.import_module(module_name)
                        plugin.module = module

                        # FIX (CRITICAL): cls(manifest.config) को try/except से सुरक्षित करें
                        # ताकि कंस्ट्रक्टर की विफलता पूरे मैनेजर को क्रैश न करे।
                        if hasattr(module, class_name):
                            cls = getattr(module, class_name)
                            try:
                                plugin.instance = cls(manifest.config)
                            except Exception as init_err:
                                plugin.module = None
                                plugin.instance = None
                                plugin.error = f"Constructor failed: {init_err}"
                                self._stats["total_failed"] += 1
                                print(
                                    f"[PluginManager] Plugin '{name}' constructor failed, "
                                    f"marking as error: {init_err}"
                                )
                                return False
                    else:
                        module = importlib.import_module(manifest.entry_point)
                        plugin.module = module
                else:
                    module = importlib.import_module(name)
                    plugin.module = module

                # मॉडल्स रजिस्टर करें
                for model in manifest.models:
                    model_id = model.get("id", "")
                    if model_id:
                        self._model_registry[model_id] = model
                        print(f"[PluginManager] Registered model: {model_id}")

                # टूल्स रजिस्टर करें
                for tool in manifest.tools:
                    tool_name = tool.get("name", "")
                    if tool_name and plugin.module:
                        tool_func = tool.get("function", "")
                        if tool_func and hasattr(plugin.module, tool_func):
                            self._tool_registry[tool_name] = getattr(plugin.module, tool_func)
                            print(f"[PluginManager] Registered tool: {tool_name}")

                # हुक्स रजिस्टर करें
                for hook in manifest.hooks:
                    if plugin.module and hasattr(plugin.module, hook):
                        self.register_hook(hook, getattr(plugin.module, hook))

                plugin.loaded_at = datetime.now().isoformat()
                plugin.error = None
                self._stats["total_loaded"] += 1

                print(f"[PluginManager] Loaded: {name} v{manifest.version}")

            except Exception as e:
                plugin.error = str(e)
                self._stats["total_failed"] += 1
                print(f"[PluginManager] Failed to load '{name}': {e}")
                return False

        # EventBus इवेंट (lock के बाहर — blocking call से deadlock से बचाव)
        if self.event_bus:
            self.event_bus.emit_sync("plugin.loaded", {
                "name": name,
                "version": manifest.version,
                "type": manifest.type,
            })

        return True

    def load_all(self) -> Dict[str, bool]:
        """
        सभी खोजे गए प्लगइन्स को लोड करता है।

        लौटाता है:
            {plugin_name: success} डिक्शनरी
        """
        results = {}
        for name in list(self._plugins.keys()):
            results[name] = self.load_plugin(name)
        return results

    def unload_plugin(self, name: str) -> bool:
        """
        एक प्लगइन को अनलोड करता है।

        पैरामीटर्स:
            name: प्लगइन का नाम

        लौटाता है:
            True अगर सफल
        """
        with self._lock:
            if name not in self._plugins:
                return False

            plugin = self._plugins[name]
            plugin_path = self.plugin_dir / name

            # मॉडल्स हटाएँ
            for model in plugin.manifest.models:
                model_id = model.get("id", "")
                self._model_registry.pop(model_id, None)

            # टूल्स हटाएँ
            for tool in plugin.manifest.tools:
                tool_name = tool.get("name", "")
                self._tool_registry.pop(tool_name, None)

            # FIX (HIGH): हुक्स हटाते समय bound methods और standalone
            # functions दोनों को सही से फ़िल्टर करें।
            # bound method  : h.__self__ is plugin.instance
            # standalone func: h.__module__ matches plugin.module.__name__
            plugin_module_name = (
                plugin.module.__name__ if plugin.module is not None else None
            )
            for hook in plugin.manifest.hooks:
                if hook in self._hook_registry:
                    kept = []
                    for h in self._hook_registry[hook]:
                        is_bound = (
                            hasattr(h, '__self__')
                            and plugin.instance is not None
                            and h.__self__ is plugin.instance
                        )
                        is_standalone = (
                            not hasattr(h, '__self__')
                            and plugin_module_name is not None
                            and getattr(h, '__module__', None) == plugin_module_name
                        )
                        if not is_bound and not is_standalone:
                            kept.append(h)
                    self._hook_registry[hook] = kept

            # FIX (STALE PATH): अनलोड होने पर प्लगइन का पाथ sys.path से हटाएँ
            # ताकि मेमोरी में कोई पुराना पाथ न बचे।
            plugin_path_str = str(plugin_path)
            if plugin_path_str in sys.path:
                try:
                    sys.path.remove(plugin_path_str)
                except ValueError:
                    pass

            # मॉड्यूल रेफरेंस हटाएँ
            plugin.module = None
            plugin.instance = None
            plugin.loaded_at = ""

            print(f"[PluginManager] Unloaded: {name}")
            return True

    # प्लगइन इनेबल/डिसेबल

    def enable_plugin(self, name: str) -> bool:
        """प्लगइन को सक्षम करें।"""
        with self._lock:
            if name in self._plugins:
                self._plugins[name].enabled = True
                return True
        return False

    def disable_plugin(self, name: str) -> bool:
        """प्लगइन को अक्षम करें।"""
        with self._lock:
            if name in self._plugins:
                self._plugins[name].enabled = False
                return True
        return False

    # हुक सिस्टम

    def register_hook(self, hook_name: str, callback: Callable):
        """
        एक नया हुक रजिस्टर करें।

        पैरामीटर्स:
            hook_name: हुक का नाम (जैसे "pre_task", "post_task", "on_error")
            callback: कॉलबैक फंक्शन
        """
        with self._lock:
            if hook_name not in self._hook_registry:
                self._hook_registry[hook_name] = []
            self._hook_registry[hook_name].append(callback)
            self._stats["hooks_registered"] += 1

    def execute_hook(self, hook_name: str, data: Optional[Dict] = None) -> List[Any]:
        """
        किसी हुक के सभी रजिस्टर्ड कॉलबैक चलाएँ।
        disabled प्लगइन के हुक्स skip किए जाते हैं।

        पैरामीटर्स:
            hook_name: हुक का नाम
            data: हुक को पास किया जाने वाला डेटा

        लौटाता है:
            सभी कॉलबैक के परिणामों की सूची
        """
        results = []
        with self._lock:
            callbacks = list(self._hook_registry.get(hook_name, []))

        for callback in callbacks:
            # FIX (HIGH): enabled == False प्लगइन के हुक्स रन नहीं होने चाहिए।
            # bound method के लिए __self__ से plugin instance खोजें;
            # standalone function के लिए __module__ से plugin खोजें।
            owner_plugin = self._find_plugin_for_callable(callback)
            if owner_plugin is not None and not owner_plugin.enabled:
                continue

            try:
                result = callback(data or {})
                results.append(result)
            except Exception as e:
                print(f"[PluginManager] Hook '{hook_name}' callback failed: {e}")
                results.append(None)

        return results

    def _find_plugin_for_callable(self, callback: Callable) -> Optional[LoadedPlugin]:
        """
        किसी callable से उसका owner LoadedPlugin खोजता है।
        bound method और standalone function दोनों को handle करता है।
        """
        with self._lock:
            for plugin in self._plugins.values():
                if plugin.instance is not None and hasattr(callback, '__self__'):
                    if callback.__self__ is plugin.instance:
                        return plugin
                if plugin.module is not None and not hasattr(callback, '__self__'):
                    if getattr(callback, '__module__', None) == plugin.module.__name__:
                        return plugin
        return None

    # मॉडल रजिस्ट्री

    def get_plugin_models(self) -> Dict[str, Dict]:
        """सभी प्लगइन-जोड़े गए मॉडल्स लौटाएँ।"""
        with self._lock:
            return dict(self._model_registry)

    def add_plugin_model(self, model_info: Dict[str, Any]) -> bool:
        """
        प्रोग्रामेटिक रूप से एक नया मॉडल जोड़ें।

        पैरामीटर्स:
            model_info: मॉडल जानकारी (id, max_tokens, strengths, etc.)

        लौटाता है:
            True अगर सफल
        """
        model_id = model_info.get("id", "")
        if not model_id:
            return False
        with self._lock:
            self._model_registry[model_id] = model_info
        print(f"[PluginManager] Added model: {model_id}")
        return True

    # टूल रजिस्ट्री

    def get_plugin_tools(self) -> Dict[str, Callable]:
        """सभी प्लगइन-जोड़े गए टूल्स लौटाएँ।"""
        with self._lock:
            return dict(self._tool_registry)

    def execute_tool(self, tool_name: str, *args, **kwargs) -> Any:
        """
        प्लगइन द्वारा जोड़ा गया टूल चलाएँ।
        disabled प्लगइन के टूल्स नहीं चलते।

        पैरामीटर्स:
            tool_name: टूल का नाम
            *args, **kwargs: टूल को पास किए जाने वाले आर्गुमेंट्स

        लौटाता है:
            टूल का परिणाम
        """
        with self._lock:
            tool = self._tool_registry.get(tool_name)

        if not tool:
            raise ValueError(f"Tool '{tool_name}' not found in plugin registry")

        # FIX (HIGH): disabled प्लगइन का टूल execute नहीं होना चाहिए।
        owner_plugin = self._find_plugin_for_callable(tool)
        if owner_plugin is not None and not owner_plugin.enabled:
            raise RuntimeError(
                f"Tool '{tool_name}' belongs to disabled plugin "
                f"'{owner_plugin.manifest.name}' and cannot be executed."
            )

        return tool(*args, **kwargs)

    # प्लगइन जानकारी

    def get_plugin_info(self, name: str) -> Optional[Dict[str, Any]]:
        """किसी प्लगइन की पूरी जानकारी लौटाएँ।"""
        with self._lock:
            plugin = self._plugins.get(name)
            if not plugin:
                return None
            return {
                "name": plugin.manifest.name,
                "version": plugin.manifest.version,
                "description": plugin.manifest.description,
                "author": plugin.manifest.author,
                "type": plugin.manifest.type,
                "enabled": plugin.enabled,
                "loaded": plugin.module is not None,
                "loaded_at": plugin.loaded_at,
                "error": plugin.error,
                "dependencies": plugin.manifest.dependencies,
                "tools_count": len(plugin.manifest.tools),
                "models_count": len(plugin.manifest.models),
            }

    def list_plugins(self) -> List[Dict[str, Any]]:
        """सभी रजिस्टर्ड प्लगइन्स की सूची लौटाएँ।"""
        return [
            self.get_plugin_info(name)
            for name in self._plugins
            if self.get_plugin_info(name) is not None
        ]

    # स्टैट्स

    def get_stats(self) -> Dict[str, Any]:
        """प्लगइन मैनेजर के आँकड़े लौटाएँ।"""
        # FIX (RACE CONDITION): stats.update({...}) ब्लॉक को with self._lock: के
        # अंदर खिसकाया — self._plugins और self._model_registry को रीड करते समय
        # 100% थ्रेड-सेफ्टी सुनिश्चित होती है।
        with self._lock:
            stats = dict(self._stats)
            stats.update({
                "plugins_registered": len(self._plugins),
                "plugins_loaded": sum(1 for p in self._plugins.values() if p.module is not None),
                "plugins_enabled": sum(1 for p in self._plugins.values() if p.enabled),
                "models_registered": len(self._model_registry),
                "tools_registered": len(self._tool_registry),
                "hooks_total": sum(len(h) for h in self._hook_registry.values()),
            })
        return stats