"""Memory provider plugin discovery. Scans four sources for memory provider plugins: 1. Bundled providers: ``plugins/memory//`` (shipped with hermes-agent) 2. User-installed providers: ``$HERMES_HOME/plugins//`` 3. Project-local providers: ``./.hermes/plugins//``, opt-in via ``HERMES_ENABLE_PROJECT_PLUGINS`` 4. Pip-installed providers: ``hermes_agent.memory_providers`` entry points Directory providers must contain ``__init__.py`` with a class implementing the MemoryProvider ABC. Pip packages expose a provider or ``register(ctx)`` callback through the entry-point group. These are the same four sources the general ``PluginManager`` scans, but the precedence is deliberately the reverse of its later-source-wins order: here **bundled wins**, then user, then project, then entry point. A memory provider is activated by name, so letting a directory dropped into the working tree shadow a shipped provider would silently redirect the agent's memory. Changing this order is a breaking change, not a cleanup. Only ONE provider can be active at a time, selected via ``memory.provider`` in config.yaml. Usage: from plugins.memory import discover_memory_providers, load_memory_provider available = discover_memory_providers() # [(name, desc, available), ...] provider = load_memory_provider("mnemosyne") # MemoryProvider instance """ from __future__ import annotations import importlib import importlib.machinery import importlib.metadata import importlib.util import logging import sys from pathlib import Path from typing import List, Optional, Tuple, TYPE_CHECKING from hermes_cli.config import cfg_get if TYPE_CHECKING: from agent.memory_provider import MemoryProvider logger = logging.getLogger(__name__) _MEMORY_PLUGINS_DIR = Path(__file__).parent ENTRY_POINTS_GROUP = "hermes_agent.memory_providers" _REGISTERED_MEMORY_PROVIDER_SKILLS: dict[str, Path] = {} # Synthetic parent package for user-installed providers, so they don't # collide with bundled providers in sys.modules. _USER_NAMESPACE = "_hermes_user_memory" def _register_synthetic_package(name: str, search_locations: List[str]) -> None: """Register an empty package shell in sys.modules. User-installed providers import as ``_hermes_user_memory.``, a dotted name whose parents exist nowhere on disk. Unless those parents are present in ``sys.modules``, any relative import inside the plugin (``from . import config``) fails with ``ModuleNotFoundError: No module named '_hermes_user_memory'`` — the same reason the loader already registers ``plugins`` and ``plugins.memory`` for bundled providers. """ if name in sys.modules: return spec = importlib.machinery.ModuleSpec(name, None, is_package=True) spec.submodule_search_locations = search_locations sys.modules[name] = importlib.util.module_from_spec(spec) # --------------------------------------------------------------------------- # Directory helpers # --------------------------------------------------------------------------- def _get_user_plugins_dir() -> Optional[Path]: """Return ``$HERMES_HOME/plugins/`` or None if unavailable.""" try: from hermes_constants import get_hermes_home d = get_hermes_home() / "plugins" return d if d.is_dir() else None except Exception: return None def _get_project_plugins_dir() -> Optional[Path]: """Return ``./.hermes/plugins/`` or None if unavailable or not opted in. Gated on ``HERMES_ENABLE_PROJECT_PLUGINS`` exactly as the general ``PluginManager`` gates its own project scan — a repository you merely ``cd`` into must not be able to offer the agent a memory backend. """ try: from hermes_cli.plugins import _env_enabled if not _env_enabled("HERMES_ENABLE_PROJECT_PLUGINS"): return None d = Path.cwd() / ".hermes" / "plugins" return d if d.is_dir() else None except Exception: return None def _is_memory_provider_dir(path: Path) -> bool: """Heuristic: does *path* look like a memory provider plugin? Checks for ``register_memory_provider`` or ``MemoryProvider`` in the ``__init__.py`` source. Cheap text scan — no import needed. """ init_file = path / "__init__.py" if not init_file.exists(): return False try: source = init_file.read_text(errors="replace", encoding="utf-8")[:8192] return "register_memory_provider" in source or "MemoryProvider" in source except Exception: return False def _iter_provider_dirs() -> List[Tuple[str, Path]]: """Yield ``(name, path)`` for all discovered provider directories. Scans bundled, then user-installed, then project-local. Bundled takes precedence on name collisions (first-seen wins via ``seen`` set). """ seen: set = set() dirs: List[Tuple[str, Path]] = [] # 1. Bundled providers (plugins/memory//) if _MEMORY_PLUGINS_DIR.is_dir(): for child in sorted(_MEMORY_PLUGINS_DIR.iterdir()): if not child.is_dir() or child.name.startswith(("_", ".")): continue if not (child / "__init__.py").exists(): continue seen.add(child.name) dirs.append((child.name, child)) # 2. User-installed providers ($HERMES_HOME/plugins//) # 3. Project-local providers (./.hermes/plugins//), opt-in for source_dir in (_get_user_plugins_dir(), _get_project_plugins_dir()): if not source_dir: continue for child in sorted(source_dir.iterdir()): if not child.is_dir() or child.name.startswith(("_", ".")): continue if child.name in seen: continue # earlier source wins if not _is_memory_provider_dir(child): continue # skip non-memory plugins seen.add(child.name) dirs.append((child.name, child)) return dirs def _iter_entry_points(): """Yield pip-installed memory provider entry points.""" try: eps = importlib.metadata.entry_points() if hasattr(eps, "select"): return list(eps.select(group=ENTRY_POINTS_GROUP)) if isinstance(eps, dict): return list(eps.get(ENTRY_POINTS_GROUP, [])) return [ep for ep in eps if ep.group == ENTRY_POINTS_GROUP] except Exception as exc: logger.debug("Memory provider entry-point scan failed: %s", exc) return [] def find_provider_dir(name: str) -> Optional[Path]: """Resolve a provider name to the directory holding its files. Checks bundled, then user-installed, then project-local, then the package directory of a pip entry-point provider. The entry-point case matters because two of a provider's files are read from disk rather than imported: ``config_schema.py`` (loaded by path so the web server never pulls in the agent runtime — see ``plugins/memory/config_schema.py``) and ``cli.py`` (loaded by ``discover_plugin_cli_commands`` at argparse time). Without a directory, a pip-installed provider silently loses its dashboard config panel and its ``hermes `` subcommands — working, but a second-class citizen next to a directory install. """ # Bundled bundled = _MEMORY_PLUGINS_DIR / name if bundled.is_dir() and (bundled / "__init__.py").exists(): return bundled # User-installed, then project-local for source_dir in (_get_user_plugins_dir(), _get_project_plugins_dir()): if not source_dir: continue candidate = source_dir / name if candidate.is_dir() and _is_memory_provider_dir(candidate): return candidate # Pip entry point return _entry_point_package_dir(find_provider_entry_point(name)) def _entry_point_package_dir(entry_point) -> Optional[Path]: """The directory of an entry point's module, resolved WITHOUT importing it. Discovery must stay free of third-party imports: ``find_provider_dir`` is called from the dashboard and from argparse setup, long before the operator has selected a provider, so importing every installed candidate would run arbitrary code on the strength of a package merely being present. ``resolve_module_origin`` walks the module's file layout instead. Only package entry points (``pkg/__init__.py``) yield a directory — a provider pointed at a bare ``module.py`` has nowhere to put a sibling ``config_schema.py``, so it correctly resolves to None. """ if entry_point is None: return None try: from hermes_cli.plugins import resolve_module_origin module_name = (entry_point.value or "").split(":")[0].strip() origin = resolve_module_origin(module_name) if not origin: return None path = Path(origin) return path.parent if path.name == "__init__.py" else None except Exception as exc: logger.debug("Could not resolve directory for entry point '%s': %s", getattr(entry_point, "name", "?"), exc) return None def find_provider_entry_point(name: str): """Resolve a provider name to a pip entry point, if installed.""" for entry_point in _iter_entry_points(): if entry_point.name == name: return entry_point return None # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def list_memory_provider_names() -> List[str]: """Cheap name-only listing of discoverable memory providers. Unlike :func:`discover_memory_providers`, this does NOT import provider modules or run availability checks — a directory scan plus entry-point *enumeration*, which reads distribution metadata without executing any of it. Safe to call at module-import time (e.g. when building the dashboard config schema, where it fills the ``memory.provider`` dropdown). """ names = {name for name, _ in _iter_provider_dirs()} names.update(ep.name for ep in _iter_entry_points()) return sorted(names) def discover_memory_providers() -> List[Tuple[str, str, bool]]: """Scan directory and pip entry-point memory providers. Returns list of (name, description, is_available) tuples. Bundled providers take precedence on name collisions, followed by user-installed directory providers, then pip entry-point providers. """ results = [] seen: set[str] = set() for name, child in _iter_provider_dirs(): # Read description from plugin.yaml if available desc = "" yaml_file = child / "plugin.yaml" if yaml_file.exists(): try: import yaml with open(yaml_file, encoding="utf-8-sig") as f: meta = yaml.safe_load(f) or {} desc = meta.get("description", "") except Exception: pass # Quick availability check — try loading and calling is_available() available = True try: provider = _load_provider_from_dir(child, register_skills=False) if provider: available = provider.is_available() else: available = False except Exception: available = False results.append((name, desc, available)) seen.add(name) for entry_point in _iter_entry_points(): name = entry_point.name if name in seen: continue desc = "" available = True try: provider = _load_provider_from_entry_point( entry_point, register_skills=False, ) if provider: available = provider.is_available() else: available = False except Exception: available = False results.append((name, desc, available)) seen.add(name) return results def load_memory_provider( name: str, *, register_skills: Optional[bool] = None, ) -> Optional["MemoryProvider"]: """Load and return a MemoryProvider instance by name. Checks bundled (``plugins/memory//``), user-installed (``$HERMES_HOME/plugins//``), and pip entry-point providers. Bundled providers take precedence on name collisions. Skills register only when *name* is the configured active provider unless ``register_skills`` is passed explicitly. This keeps status and setup inspection of inactive providers free of registry side effects. Returns None if the provider is not found or fails to load. """ if register_skills is None: register_skills = name == _get_active_memory_provider() provider_dir = find_provider_dir(name) entry_point = None if provider_dir else find_provider_entry_point(name) if not provider_dir and entry_point is None: logger.debug( "Memory provider '%s' not found in bundled, user plugins, or entry points", name, ) return None try: provider = ( _load_provider_from_dir(provider_dir, register_skills=register_skills) if provider_dir else _load_provider_from_entry_point( entry_point, register_skills=register_skills, ) ) if provider: return provider logger.warning("Memory provider '%s' loaded but no provider instance found", name) return None except Exception as e: logger.warning("Failed to load memory provider '%s': %s", name, e) return None def _load_provider_from_entry_point( entry_point, *, register_skills: bool = True, ) -> Optional["MemoryProvider"]: """Import a provider entry point and extract the MemoryProvider instance.""" from agent.memory_provider import MemoryProvider loaded = entry_point.load() if isinstance(loaded, MemoryProvider): return loaded if isinstance(loaded, type) and issubclass(loaded, MemoryProvider): try: return loaded() except Exception: pass if hasattr(loaded, "register"): collector = _ProviderCollector(entry_point.name, register_skills=register_skills) loaded.register(collector) if collector.provider: return collector.provider if callable(loaded): try: provider = loaded() if isinstance(provider, MemoryProvider): return provider except TypeError: pass collector = _ProviderCollector(entry_point.name, register_skills=register_skills) loaded(collector) return collector.provider for attr_name in dir(loaded): attr = getattr(loaded, attr_name, None) if (isinstance(attr, type) and issubclass(attr, MemoryProvider) and attr is not MemoryProvider): try: return attr() except Exception: pass logger.debug("Memory provider entry point '%s' loaded no provider", entry_point.name) return None def _load_provider_from_dir( provider_dir: Path, *, register_skills: bool = True, ) -> Optional["MemoryProvider"]: """Import a provider module and extract the MemoryProvider instance. The module must have either: - A register(ctx) function (plugin-style) — we simulate a ctx - A top-level class that extends MemoryProvider — we instantiate it """ name = provider_dir.name # Use a separate namespace for user-installed plugins so they don't # collide with bundled providers in sys.modules. _is_bundled = _MEMORY_PLUGINS_DIR in provider_dir.parents or provider_dir.parent == _MEMORY_PLUGINS_DIR module_name = f"plugins.memory.{name}" if _is_bundled else f"{_USER_NAMESPACE}.{name}" init_file = provider_dir / "__init__.py" if not init_file.exists(): return None # Check if already loaded. A synthetic package shell registered by # discover_plugin_cli_commands() for relative-import support has no # __file__; only reuse modules that were actually loaded from disk. cached = sys.modules.get(module_name) if cached is not None and getattr(cached, "__file__", None): mod = cached else: # Handle relative imports within the plugin # First ensure the parent packages are registered for parent in ("plugins", "plugins.memory"): if parent not in sys.modules: parent_path = Path(__file__).parent if parent == "plugins": parent_path = parent_path.parent parent_init = parent_path / "__init__.py" if parent_init.exists(): spec = importlib.util.spec_from_file_location( parent, str(parent_init), submodule_search_locations=[str(parent_path)] ) if spec: parent_mod = importlib.util.module_from_spec(spec) sys.modules[parent] = parent_mod try: spec.loader.exec_module(parent_mod) except Exception: pass # User-installed plugins need their synthetic parent registered the # same way, or relative imports inside the plugin cannot resolve. if not _is_bundled: _register_synthetic_package(_USER_NAMESPACE, []) # Now load the provider module spec = importlib.util.spec_from_file_location( module_name, str(init_file), submodule_search_locations=[str(provider_dir)] ) if not spec: return None mod = importlib.util.module_from_spec(spec) sys.modules[module_name] = mod # Register submodules so relative imports work # e.g., "from .store import MemoryStore" in holographic plugin for sub_file in provider_dir.glob("*.py"): if sub_file.name == "__init__.py": continue sub_name = sub_file.stem full_sub_name = f"{module_name}.{sub_name}" if full_sub_name not in sys.modules: sub_spec = importlib.util.spec_from_file_location( full_sub_name, str(sub_file) ) if sub_spec: sub_mod = importlib.util.module_from_spec(sub_spec) sys.modules[full_sub_name] = sub_mod try: sub_spec.loader.exec_module(sub_mod) except Exception as e: logger.debug("Failed to load submodule %s: %s", full_sub_name, e) try: spec.loader.exec_module(mod) except Exception as e: logger.debug("Failed to exec_module %s: %s", module_name, e) sys.modules.pop(module_name, None) return None # Try register(ctx) pattern first (how our plugins are written) if hasattr(mod, "register"): collector = _ProviderCollector(name, register_skills=register_skills) try: mod.register(collector) except Exception as e: # A raise AFTER register_memory_provider() must not cost us the # provider. Falling through to the subclass scan below would # discard the instance the plugin configured and hand back a bare # second one — a silent downgrade that looks like success. if collector.provider is None: logger.debug("register() failed for %s: %s", name, e) else: logger.warning( "Memory provider '%s' raised after registering (%s) — " "using the registered provider; later registrations were skipped", name, e, ) if collector.provider: return collector.provider # Fallback: find a MemoryProvider subclass and instantiate it from agent.memory_provider import MemoryProvider for attr_name in dir(mod): attr = getattr(mod, attr_name, None) if (isinstance(attr, type) and issubclass(attr, MemoryProvider) and attr is not MemoryProvider): try: return attr() except Exception: pass return None class _ProviderCollector: """Plugin context for memory providers. Captures ``register_memory_provider`` directly — that is the one call the exclusive activation path owns — and delegates everything else to a real ``PluginContext`` (see ``__getattr__``), so a memory provider has the same registration surface as any other plugin. """ def __init__(self, name: str, *, register_skills: bool = True): self.name = name self.provider = None self._register_skills = register_skills self._context = None def register_memory_provider(self, provider): self.provider = provider def register_skill(self, *args, **kwargs): """Forward plugin-provided skills to the general plugin registry. Handled explicitly rather than through ``__getattr__`` because skills are tracked for pruning: switching the active provider has to retract the skills the previous one registered, which needs the qualified name and resolved path recorded here. Gated on ``register_skills`` so merely *inspecting* an inactive provider — ``hermes memory status``, the setup picker — leaves no registry side effects behind. """ if not self._register_skills: return try: manager_context = self._plugin_context() manager_context.register_skill(*args, **kwargs) skill_name = args[0] if args else kwargs.get("name") qualified_name = f"{self.name}:{skill_name}" from hermes_cli.plugins import get_plugin_manager registered_path = get_plugin_manager().find_plugin_skill(qualified_name) if registered_path is not None: _REGISTERED_MEMORY_PROVIDER_SKILLS[qualified_name] = registered_path except Exception as exc: logger.debug("Memory provider '%s' failed to register skill: %s", self.name, exc) def register_cli_command(self, *args, **kwargs): pass # CLI registration happens via discover_plugin_cli_commands() def __getattr__(self, attr: str): """Delegate any other ``register_*`` call to a real ``PluginContext``. Memory providers used to get a hand-maintained stub of three no-ops here, which had two failure modes. Calls it *did* know about (``register_tool``, ``register_hook``) were silently dropped, so a provider's tools simply never appeared. Calls it did *not* know about raised ``AttributeError`` — and ``register_auxiliary_task`` is one of them, despite ``PluginContext.register_auxiliary_task`` documenting a memory provider (hindsight's pre-retain dedup) as its worked example. That exception surfaces as "register() failed" and costs the provider. Delegating instead of enumerating means this can never drift behind ``PluginContext`` again: a capability added there works for memory providers on the same commit, which is what the "widen the generic plugin surface" rule in AGENTS.md asks for. Only ``register_*`` is forwarded. Everything else raises normally, so a typo still fails loudly rather than being absorbed. """ if not attr.startswith("register_"): raise AttributeError(attr) def _forward(*args, **kwargs): try: return self._plugin_context().__getattribute__(attr)(*args, **kwargs) except Exception as exc: # A secondary registration must not cost the provider itself — # by the time these run, register_memory_provider has usually # already handed us the instance the agent needs. logger.warning( "Memory provider '%s' failed to %s: %s", self.name, attr, exc ) return None return _forward def _plugin_context(self): """A real ``PluginContext`` for this provider, built once on demand. Lazy because the common case — a provider that only calls ``register_memory_provider`` — must not pay for importing the general plugin manager, which discovery touches on every hermes startup. """ if self._context is None: from hermes_cli.plugins import PluginContext, PluginManifest, get_plugin_manager manifest = PluginManifest(name=self.name, key=self.name) self._context = PluginContext(manifest, get_plugin_manager()) return self._context def _get_active_memory_provider() -> Optional[str]: """Read the active memory provider name from config.yaml. Returns the provider name (e.g. ``"honcho"``) or None if no external provider is configured. Lightweight — only reads config, no plugin loading. """ try: from hermes_cli.config import load_config config = load_config() return cfg_get(config, "memory", "provider") or None except Exception: return None def _prune_inactive_memory_provider_skills( active_provider: Optional[str] = None, ) -> None: """Remove tracked skills that no longer belong to the active provider.""" if active_provider is None: active_provider = _get_active_memory_provider() from hermes_cli.plugins import get_plugin_manager manager = get_plugin_manager() for qualified_name, registered_path in list( _REGISTERED_MEMORY_PROVIDER_SKILLS.items() ): namespace, _, _ = qualified_name.partition(":") if namespace == active_provider: continue if manager.find_plugin_skill(qualified_name) == registered_path: manager.remove_plugin_skill(qualified_name) _REGISTERED_MEMORY_PROVIDER_SKILLS.pop(qualified_name, None) def discover_plugin_cli_commands() -> List[dict]: """Return CLI commands for the **active** memory plugin only. Only one memory provider can be active at a time (set via ``memory.provider`` in config.yaml). This function reads that value and only loads CLI registration for the matching plugin. If no provider is active, no commands are registered. Looks for a ``register_cli(subparser)`` function in the active plugin's ``cli.py``. Returns a list of at most one dict with keys: ``name``, ``help``, ``description``, ``setup_fn``, ``handler_fn``. This is a lightweight scan — it only imports ``cli.py``, not the full plugin module. Safe to call during argparse setup before any provider is loaded. """ results: List[dict] = [] if not _MEMORY_PLUGINS_DIR.is_dir(): return results active_provider = _get_active_memory_provider() if not active_provider: return results # Only look at the active provider's directory plugin_dir = find_provider_dir(active_provider) if not plugin_dir: return results cli_file = plugin_dir / "cli.py" if not cli_file.exists(): return results _is_bundled = _MEMORY_PLUGINS_DIR in plugin_dir.parents or plugin_dir.parent == _MEMORY_PLUGINS_DIR module_name = f"plugins.memory.{active_provider}.cli" if _is_bundled else f"{_USER_NAMESPACE}.{active_provider}.cli" try: # Import the CLI module (lightweight — no SDK needed) if module_name in sys.modules: cli_mod = sys.modules[module_name] else: if not _is_bundled: # cli.py imports as _hermes_user_memory..cli, usually # before the provider itself is loaded. Register its parent # packages so relative imports inside cli.py # ("from . import config") resolve without executing the # plugin's __init__.py. The package shell has no __file__, # so _load_provider_from_dir() will still load the real # module later instead of reusing the shell. _register_synthetic_package(_USER_NAMESPACE, []) _register_synthetic_package( f"{_USER_NAMESPACE}.{active_provider}", [str(plugin_dir)] ) spec = importlib.util.spec_from_file_location( module_name, str(cli_file) ) if not spec or not spec.loader: return results cli_mod = importlib.util.module_from_spec(spec) sys.modules[module_name] = cli_mod spec.loader.exec_module(cli_mod) register_cli = getattr(cli_mod, "register_cli", None) if not callable(register_cli): return results # Read metadata from plugin.yaml if available help_text = f"Manage {active_provider} memory plugin" description = "" yaml_file = plugin_dir / "plugin.yaml" if yaml_file.exists(): try: import yaml with open(yaml_file, encoding="utf-8-sig") as f: meta = yaml.safe_load(f) or {} desc = meta.get("description", "") if desc: help_text = desc description = desc except Exception: pass handler_fn = getattr(cli_mod, f"{active_provider}_command", None) or \ getattr(cli_mod, "honcho_command", None) results.append({ "name": active_provider, "help": help_text, "description": description, "setup_fn": register_cli, "handler_fn": handler_fn, "plugin": active_provider, }) except Exception as e: logger.debug("Failed to scan CLI for memory plugin '%s': %s", active_provider, e) return results