| """Context engine plugin discovery: ``plugins/context_engine/<name>/`` → ``ContextEngine``. |
| Engines ship in the repo, separate from the general plugin system; only one is active |
| (``context.engine`` in config.yaml; default ``"compressor"``, the built-in ContextCompressor).""" |
|
|
| from __future__ import annotations |
|
|
| import logging |
| from pathlib import Path |
| from typing import List, Optional, Tuple |
|
|
| from plugins import plugin_loader as _loader |
|
|
| logger = logging.getLogger(__name__) |
|
|
| _CONTEXT_ENGINE_PLUGINS_DIR = Path(__file__).parent |
|
|
|
|
| def discover_context_engines() -> List[Tuple[str, str, bool]]: |
| """Return ``[(name, description, is_available), ...]`` for every bundled engine.""" |
| return [(child.name, _loader.read_plugin_description(child), |
| _loader.probe_availability(lambda c=child: _load_engine_from_dir(c))) |
| for child in _loader.iter_plugin_dirs(_CONTEXT_ENGINE_PLUGINS_DIR)] |
|
|
|
|
| def load_context_engine(name: str) -> Optional["ContextEngine"]: |
| """Load a ContextEngine instance by name; None if not found or it fails to load.""" |
| engine_dir = _CONTEXT_ENGINE_PLUGINS_DIR / name |
| if not engine_dir.is_dir(): |
| logger.debug("Context engine '%s' not found in %s", name, _CONTEXT_ENGINE_PLUGINS_DIR) |
| return None |
| return _loader.load_named( |
| name, engine_dir, _load_engine_from_dir, kind="Context engine", noun="engine", logger=logger |
| ) |
|
|
|
|
| def _load_engine_from_dir(engine_dir: Path) -> Optional["ContextEngine"]: |
| """Import an engine module and extract its ContextEngine (register(ctx) or subclass).""" |
| from agent.context_engine import ContextEngine |
| name = engine_dir.name |
| mod = _loader.load_plugin_module( |
| f"plugins.context_engine.{name}", engine_dir, |
| parents=("plugins", "plugins.context_engine"), logger=logger) |
| return mod and _loader.instance_from_module( |
| mod, collector=_EngineCollector(engine_name=name), collected_attr="engine", |
| base_cls=ContextEngine, name=name, logger=logger) |
|
|
|
|
| class _EngineCollector(_loader.NoopPluginContext): |
| """Captures register_context_engine; forwards register_command to the global plugin command |
| registry so engine slash commands behave like plugin ones.""" |
|
|
| def __init__(self, engine_name: str = ""): |
| self.engine = None |
| self._engine_name = engine_name or "context_engine" |
|
|
| def register_context_engine(self, engine): |
| self.engine = engine |
|
|
| def register_command(self, name: str, handler, description: str = "", args_hint: str = "") -> None: |
| clean = (name or "").lower().strip().lstrip("/").replace(" ", "-") |
| if not clean: |
| logger.warning("Context engine '%s' tried to register a command with an empty name.", |
| self._engine_name) |
| return |
| conflict = "Context engine '%s' tried to register command '/%s' which %s Skipping." |
| try: |
| from hermes_cli.commands import resolve_command |
| if resolve_command(clean) is not None: |
| logger.warning(conflict, self._engine_name, clean, "conflicts with a built-in command.") |
| return |
| except Exception: |
| pass |
| try: |
| from hermes_cli.plugins import get_plugin_manager |
| manager = get_plugin_manager() |
| if clean in manager._plugin_commands: |
| logger.warning(conflict, self._engine_name, clean, "is already registered by a plugin.") |
| return |
| manager._plugin_commands[clean] = { |
| "handler": handler, "description": description or "Context engine command", |
| "plugin": f"context-engine:{self._engine_name}", "args_hint": (args_hint or "").strip()} |
| logger.debug("Context engine '%s' registered command: /%s", self._engine_name, clean) |
| except Exception as exc: |
| logger.debug("Context engine '%s' could not register /%s: %s", self._engine_name, clean, exc) |
|
|
|
|
| |
| |
| |
| |
| import importlib.util |
| import sys |
| |
|
|
