"""OMEGA Plugin Interface — extensible discovery for additional modules. Core provides entry-point-based plugin discovery. Extension packages register via ``[project.entry-points."omega.plugins"]`` in their pyproject.toml. """ from __future__ import annotations import logging from typing import Any, Callable logger = logging.getLogger("omega.plugins") class OmegaPlugin: """Base class for OMEGA plugins. Subclasses should populate these class-level attributes: - ``TOOL_SCHEMAS``: list of MCP tool schema dicts - ``HANDLERS``: dict mapping tool name → async handler function - ``HOOK_HANDLERS``: dict mapping hook name → sync handler function - ``CLI_COMMANDS``: list of (name, setup_func) tuples where setup_func(subparsers) registers an argparse subparser; the plugin class should also provide a ``cmd_{name}(args)`` method as the command handler - ``HOOKS_JSON``: dict matching the hooks.json manifest format (optional) - ``RETRIEVAL_PROFILES``: dict mapping event_type → (vec, text, word, ctx, graph) phase weights - ``SCORE_MODIFIERS``: list of fn(node_id, score, metadata) → score callables """ TOOL_SCHEMAS: list[dict[str, Any]] = [] HANDLERS: dict[str, Callable] = {} HOOK_HANDLERS: dict[str, Callable] = {} CLI_COMMANDS: list[tuple[str, Callable]] = [] HOOKS_JSON: dict[str, Any] = {} RETRIEVAL_PROFILES: dict[str, tuple] = {} SCORE_MODIFIERS: list[Callable] = [] def discover_plugins() -> list[OmegaPlugin]: """Discover and instantiate all registered OMEGA plugins. Looks up ``omega.plugins`` entry-point group via importlib.metadata. Each entry point should reference a class that inherits from OmegaPlugin. Returns an empty list if no plugins are installed. """ plugins: list[OmegaPlugin] = [] try: from importlib.metadata import entry_points eps = entry_points(group="omega.plugins") for ep in eps: try: plugin_cls = ep.load() if isinstance(plugin_cls, type) and issubclass(plugin_cls, OmegaPlugin): plugins.append(plugin_cls()) elif isinstance(plugin_cls, OmegaPlugin): plugins.append(plugin_cls) else: logger.warning("Plugin %s is not an OmegaPlugin subclass, skipping", ep.name) except Exception as e: logger.warning("Failed to load plugin %s: %s", ep.name, e) except Exception as e: logger.debug("Plugin discovery unavailable: %s", e) return plugins