File size: 4,414 Bytes
674d56a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """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"]: # noqa: F821
"""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"]: # noqa: F821
"""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)
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.
import importlib.util # noqa: F401,E402
import sys # noqa: F401,E402
# ---- END PLUGIN-COMPAT ----
|