339229490c
# Conflicts: # .github/workflows/tests-os.yml # .github/workflows/tests.yml # Dockerfile # hermes_cli/memory_setup.py # hermes_cli/update_cmd.py # hermes_cli/web_server_memory.py # plugins/memory/hindsight/README.md # plugins/memory/hindsight/__init__.py # plugins/memory/hindsight/embedded.py # plugins/memory/hindsight/plugin.yaml # plugins/memory/hindsight/settings.py # plugins/memory/hindsight/setup.py # tests/plugins/memory/test_bom_tolerant_config_reads.py # tests/plugins/memory/test_hindsight_env_perms.py # tests/plugins/memory/test_hindsight_provider.py # tests/tools/test_lazy_deps.py # tools/lazy_deps.py # uv.lock # website/docs/user-guide/docker.md # website/docs/user-guide/features/memory-providers.md
533 lines
24 KiB
Python
533 lines
24 KiB
Python
"""Memory provider plugin discovery: bundled ``plugins/memory/<name>/``, user
|
|
``$HERMES_HOME/plugins/<name>/``, project ``./.hermes/plugins/<name>/`` (opt-in via
|
|
HERMES_ENABLE_PROJECT_PLUGINS), then ``hermes_agent.memory_providers`` entry points.
|
|
Precedence is deliberately the REVERSE of PluginManager's later-source-wins:
|
|
bundled wins, then user, project, entry point — a provider is activated by name
|
|
(``memory.provider``, one at a time), so a directory dropped into the working tree
|
|
must never shadow a shipped provider. Changing this order is a breaking change.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.metadata
|
|
import importlib.util
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import List, Optional, Tuple, TYPE_CHECKING
|
|
|
|
from hermes_cli.config import cfg_get
|
|
from plugins import plugin_loader as _loader
|
|
|
|
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"
|
|
# Per Hermes home (plugin managers are per home too): pruning under one multiplexed profile must
|
|
# only retract that profile's provider skills, never a sibling profile's.
|
|
_REGISTERED_MEMORY_PROVIDER_SKILLS: dict[str, dict[str, Path]] = {}
|
|
# Native extensions whose first import must not race another thread (#58083 warm-up).
|
|
_NATIVE_WARM_IMPORTS: Tuple[str, ...] = ("numpy",)
|
|
|
|
|
|
def _registered_skills_for_active_home() -> dict[str, Path]:
|
|
from hermes_constants import hermes_home_key
|
|
|
|
return _REGISTERED_MEMORY_PROVIDER_SKILLS.setdefault(hermes_home_key(), {})
|
|
|
|
# Synthetic parent package so user-installed providers don't collide with bundled ones.
|
|
_USER_NAMESPACE = "_hermes_user_memory"
|
|
|
|
_register_synthetic_package = _loader.register_synthetic_package
|
|
_get_user_plugins_dir = _loader.user_plugins_dir
|
|
|
|
|
|
def _get_project_plugins_dir() -> Optional[Path]:
|
|
"""``./.hermes/plugins/`` or None. Gated on HERMES_ENABLE_PROJECT_PLUGINS like the
|
|
PluginManager scan: a repo you merely ``cd`` into must not offer 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:
|
|
"""Cheap text heuristic (no import): ``__init__.py`` mentions the memory provider contract."""
|
|
init_file = path / "__init__.py"
|
|
try:
|
|
if not init_file.exists():
|
|
return False
|
|
source = init_file.read_text(errors="replace", encoding="utf-8-sig")[:8192]
|
|
return "register_memory_provider" in source or "MemoryProvider" in source
|
|
except OSError as exc: # one mode-000 / ACL-denied child must not abort discovery
|
|
logger.warning("Skipping unreadable plugin directory %s: %s", path, exc)
|
|
return False
|
|
|
|
|
|
def _is_bundled(provider_dir: Path) -> bool:
|
|
return _MEMORY_PLUGINS_DIR in provider_dir.parents or provider_dir.parent == _MEMORY_PLUGINS_DIR
|
|
|
|
|
|
def _module_name(provider_dir: Path, name: str) -> str:
|
|
"""``plugins.memory.<name>`` for bundled providers, else under the synthetic user namespace."""
|
|
if _is_bundled(provider_dir):
|
|
return f"plugins.memory.{name}"
|
|
# Separate package trees, including relative imports, across homes/sources.
|
|
digest = hashlib.sha256(str(provider_dir.resolve()).encode()).hexdigest()[:16]
|
|
return f"{_USER_NAMESPACE}.{name}__source_{digest}"
|
|
|
|
|
|
def _external_source_dirs() -> List[Path]:
|
|
"""User then project plugin roots that exist (precedence order)."""
|
|
return [d for d in (_get_user_plugins_dir(), _get_project_plugins_dir()) if d]
|
|
|
|
|
|
def _iter_provider_dirs() -> List[Tuple[str, Path]]:
|
|
"""``(name, path)`` for bundled, then user-installed, then project-local; first-seen wins."""
|
|
dirs = [(child.name, child) for child in _loader.iter_plugin_dirs(_MEMORY_PLUGINS_DIR)]
|
|
seen = {name for name, _ in dirs}
|
|
for source_dir in _external_source_dirs():
|
|
for child in sorted(source_dir.iterdir()):
|
|
if (
|
|
child.is_dir()
|
|
and not child.name.startswith(("_", "."))
|
|
and child.name not in seen
|
|
and _is_memory_provider_dir(child)
|
|
):
|
|
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))
|
|
return list(eps.get(ENTRY_POINTS_GROUP, [])) if isinstance(eps, dict) else [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]:
|
|
"""Provider name -> directory: bundled, user, project, then a pip entry point's
|
|
package dir. The entry-point case matters because ``config_schema.py`` and
|
|
``cli.py`` are read from disk, not imported; without a directory a pip-installed
|
|
provider silently loses its dashboard panel and ``hermes <provider>`` commands."""
|
|
bundled = _MEMORY_PLUGINS_DIR / name
|
|
if bundled.is_dir() and (bundled / "__init__.py").exists():
|
|
return bundled
|
|
for source_dir in _external_source_dirs():
|
|
candidate = source_dir / name
|
|
if candidate.is_dir() and _is_memory_provider_dir(candidate):
|
|
return candidate
|
|
return _entry_point_package_dir(find_provider_entry_point(name))
|
|
|
|
|
|
def _entry_point_package_dir(entry_point) -> Optional[Path]:
|
|
"""Directory of an entry point's module, resolved WITHOUT importing it (discovery
|
|
runs from the dashboard/argparse before a provider is selected; importing every
|
|
candidate would run arbitrary code). Only package entry points yield a
|
|
directory — a bare ``module.py`` has nowhere for a sibling ``config_schema.py``."""
|
|
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."""
|
|
return next((ep for ep in _iter_entry_points() if ep.name == name), None)
|
|
|
|
|
|
def list_memory_provider_names() -> List[str]:
|
|
"""Cheap name-only listing (directory scan + entry-point enumeration, no import or
|
|
availability check) — safe at module-import time (dashboard 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]]:
|
|
"""``[(name, description, is_available), ...]``; bundled wins on name collisions,
|
|
then user directories, then pip."""
|
|
results = [
|
|
(name, _loader.read_plugin_description(child),
|
|
_loader.probe_availability(lambda c=child: _load_provider_from_dir(c, register_skills=False)))
|
|
for name, child in _iter_provider_dirs()
|
|
]
|
|
seen = {name for name, _, _ in results}
|
|
for entry_point in _iter_entry_points():
|
|
if entry_point.name not in seen:
|
|
seen.add(entry_point.name)
|
|
results.append((entry_point.name, "", _loader.probe_availability(
|
|
lambda ep=entry_point: _load_provider_from_entry_point(ep, register_skills=False)
|
|
)))
|
|
return results
|
|
|
|
|
|
def load_memory_provider(name: str, *, register_skills: Optional[bool] = None) -> Optional["MemoryProvider"]:
|
|
"""Load a MemoryProvider by name (bundled, user, project, then pip entry point);
|
|
None if not found or failing to load. Skills register only for the configured
|
|
active provider unless ``register_skills`` is explicit, so inspecting inactive
|
|
providers leaves no registry side effects."""
|
|
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
|
|
if provider_dir is not None and _explicitly_disabled(name, provider_dir):
|
|
# The Plugins hub / `hermes plugins disable` park a user-installed provider in
|
|
# ``plugins.disabled``; the loader must honour it or "disabled" is a lie in the UI.
|
|
logger.warning("Memory provider '%s' is disabled via plugins.disabled; run `hermes plugins enable %s` "
|
|
"or change memory.provider.", name, name)
|
|
return None
|
|
|
|
def _load(_dir):
|
|
if provider_dir:
|
|
return _load_provider_from_dir(provider_dir, register_skills=register_skills)
|
|
return _load_provider_from_entry_point(entry_point, register_skills=register_skills)
|
|
|
|
return _loader.load_named(name, provider_dir, _load, kind="Memory provider", noun="provider", logger=logger)
|
|
|
|
|
|
def import_memory_provider_module(name: Optional[str] = None) -> bool:
|
|
"""Import the provider's module (default: the configured ``memory.provider``) WITHOUT
|
|
constructing a provider — the later ``load_memory_provider`` then hits ``sys.modules``
|
|
instead of a fresh native extension load. Exists so ``hermes acp`` can pay the heavy
|
|
import (numpy / ML stack) on the main thread before any other thread starts: on Windows
|
|
a first-time native import racing another thread's import chain deadlocked
|
|
``session/new`` (#58083). False when no provider is configured, the provider is
|
|
unknown or its import fails (agent init reports that)."""
|
|
name = name or _get_active_memory_provider()
|
|
if not name:
|
|
return False
|
|
imported = False
|
|
try:
|
|
if provider_dir := find_provider_dir(name):
|
|
imported = _load_package(provider_dir, name) is not None
|
|
elif (entry_point := find_provider_entry_point(name)) is not None:
|
|
entry_point.load()
|
|
imported = True
|
|
except Exception:
|
|
logger.debug("memory provider '%s' warm-up import failed", name, exc_info=True)
|
|
if imported:
|
|
# The deadlock is numpy's lazy ``_core`` init; embedding-backed providers (e.g. the
|
|
# hindsight plugin) defer that import to ``is_available()`` (sentence_transformers), so
|
|
# the provider module alone leaves it unwarmed. Every reporter's workaround was a plain ``import numpy`` up front.
|
|
for module in _NATIVE_WARM_IMPORTS:
|
|
try:
|
|
importlib.import_module(module)
|
|
except Exception:
|
|
logger.debug("warm-up import of %s skipped", module, exc_info=True)
|
|
return imported
|
|
|
|
|
|
def _load_package(provider_dir: Path, name: str):
|
|
"""Import the provider package at *provider_dir* under the module name the loader owns
|
|
(``plugins.memory.<name>`` bundled, synthetic user namespace otherwise); None on failure."""
|
|
return _loader.load_plugin_module(
|
|
_module_name(provider_dir, name), provider_dir,
|
|
parents=("plugins", "plugins.memory"),
|
|
logger=logger,
|
|
synthetic_namespace=None if _is_bundled(provider_dir) else _USER_NAMESPACE,
|
|
)
|
|
|
|
|
|
def import_provider_module(name: str, submodule: Optional[str] = None):
|
|
"""The package (or ``<package>.<submodule>``) of whichever copy of provider *name* is installed.
|
|
|
|
Host-side code (dashboard host-block storage, OAuth routes, doctor, profile clone) used to
|
|
``import plugins.memory.<name>.<submodule>``, which only exists for the bundled copy; a
|
|
catalog install under ``$HERMES_HOME/plugins/`` loads under the synthetic user namespace,
|
|
so those surfaces 500'd/404'd the moment the bundled copy left core. Resolving through
|
|
``find_provider_dir`` makes bundled and user-dir copies behave identically. Raises
|
|
``ImportError`` when the provider is not installed or lacks the submodule.
|
|
"""
|
|
provider_dir = find_provider_dir(name)
|
|
if provider_dir is None:
|
|
raise ImportError(f"memory provider {name!r} is not installed")
|
|
package = _load_package(provider_dir, name)
|
|
if package is None:
|
|
raise ImportError(f"memory provider {name!r} failed to import")
|
|
return importlib.import_module(f"{package.__name__}.{submodule}") if submodule else package
|
|
|
|
|
|
def _instantiate_subclass(namespace) -> Optional["MemoryProvider"]:
|
|
"""First instantiable ``MemoryProvider`` subclass found among *namespace*'s attributes."""
|
|
from agent.memory_provider import MemoryProvider
|
|
|
|
for attr_name in dir(namespace):
|
|
attr = getattr(namespace, attr_name, None)
|
|
if isinstance(attr, type) and issubclass(attr, MemoryProvider) and attr is not MemoryProvider:
|
|
try:
|
|
return attr()
|
|
except Exception:
|
|
pass
|
|
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: an
|
|
instance, a subclass, a module with ``register(ctx)``, a factory / ``register``
|
|
callable, or a namespace holding a subclass — in that order."""
|
|
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)
|
|
collector.collect(loaded.register, source=getattr(loaded, "__file__", None))
|
|
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)
|
|
collector.collect(loaded)
|
|
return collector.provider
|
|
|
|
provider = _instantiate_subclass(loaded)
|
|
if provider is None:
|
|
logger.debug("Memory provider entry point '%s' loaded no provider", entry_point.name)
|
|
return provider
|
|
|
|
|
|
def _load_provider_from_dir(provider_dir: Path, *, register_skills: bool = True) -> Optional["MemoryProvider"]:
|
|
"""Import a provider module; ``register(ctx)`` first, else a top-level subclass."""
|
|
name = provider_dir.name
|
|
mod = _load_package(provider_dir, name)
|
|
if mod is None:
|
|
return None
|
|
|
|
if hasattr(mod, "register"):
|
|
collector = _ProviderCollector(name, register_skills=register_skills)
|
|
try:
|
|
collector.collect(mod.register, source=mod.__file__)
|
|
except Exception as e:
|
|
# A raise AFTER register_memory_provider() must not cost us the provider:
|
|
# falling through to the subclass scan would hand back a bare second
|
|
# instance — 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
|
|
|
|
return _instantiate_subclass(mod)
|
|
|
|
|
|
class _ProviderCollector:
|
|
"""Plugin context for memory providers: captures ``register_memory_provider``
|
|
(the one call the activation path owns) and delegates other ``register_*``
|
|
calls to a real ``PluginContext`` so providers have the full plugin surface."""
|
|
|
|
def __init__(self, name: str, *, register_skills: bool = True):
|
|
self.name = name
|
|
self.provider = None
|
|
self._register_skills = register_skills
|
|
self._context = None
|
|
self._hook_source = None
|
|
|
|
def collect(self, register, *, source=None):
|
|
"""Run ``register`` with this collector; hooks it registers form the fallback group that
|
|
general discovery of the same source replaces (see ``PluginLedgerMixin``)."""
|
|
from hermes_cli.plugins_ledger import _hook_source_of
|
|
|
|
module = sys.modules.get(getattr(register, "__module__", ""))
|
|
self._hook_source = _hook_source_of(self.name, SimpleNamespace(__file__=source) if source else module)
|
|
manager = self._plugin_context()._manager
|
|
with manager._discovery_lock:
|
|
manager._drop_fallback_hooks(self._hook_source)
|
|
register(self)
|
|
|
|
def register_hook(self, hook_name, callback):
|
|
context = self._plugin_context()
|
|
with context._manager._discovery_lock:
|
|
return context._manager._register_fallback_hook(context, self._hook_source, hook_name, callback)
|
|
|
|
def register_memory_provider(self, provider):
|
|
self.provider = provider
|
|
|
|
def register_skill(self, *args, **kwargs):
|
|
"""Forward skills to the plugin registry, tracking qualified name + path so
|
|
switching the active provider can retract the previous one's skills. Gated
|
|
on ``register_skills`` so inspecting an inactive provider has no side effects."""
|
|
if not self._register_skills:
|
|
return
|
|
try:
|
|
self._plugin_context().register_skill(*args, **kwargs)
|
|
qualified_name = f"{self.name}:{args[0] if args else kwargs.get('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_skills_for_active_home()[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`` (a
|
|
hand-maintained stub drifted: unknown calls raised and cost the provider).
|
|
Non-``register_*`` attributes raise normally so a typo still fails loudly."""
|
|
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.
|
|
logger.warning("Memory provider '%s' failed to %s: %s", self.name, attr, exc)
|
|
return None
|
|
|
|
return _forward
|
|
|
|
def _plugin_context(self):
|
|
"""A real ``PluginContext``, built once on demand: the common provider that only
|
|
calls ``register_memory_provider`` must not pay for importing the plugin manager."""
|
|
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]:
|
|
"""Active provider name from config.yaml (``memory.provider``), or None. Reads config only."""
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
config = load_config()
|
|
return cfg_get(config, "memory", "provider") or None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _explicitly_disabled(name: str, provider_dir: Path) -> bool:
|
|
"""True when a NON-bundled provider is parked in ``plugins.disabled`` under any spelling the
|
|
plugin commands write: the provider name, its directory name or its manifest ``name:``."""
|
|
if _MEMORY_PLUGINS_DIR in provider_dir.parents:
|
|
return False
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
disabled = cfg_get(load_config(), "plugins", "disabled")
|
|
except Exception:
|
|
return False
|
|
if not isinstance(disabled, list):
|
|
return False
|
|
names = {name, provider_dir.name}
|
|
try:
|
|
import hermes_yaml as yaml
|
|
with open(provider_dir / "plugin.yaml", encoding="utf-8-sig") as f:
|
|
names.add(str((yaml.safe_load(f) or {}).get("name") or ""))
|
|
except Exception:
|
|
pass
|
|
return bool(names & {v for v in disabled if isinstance(v, str)})
|
|
|
|
|
|
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()
|
|
registered = _registered_skills_for_active_home()
|
|
for qualified_name, registered_path in list(registered.items()):
|
|
if qualified_name.partition(":")[0] == active_provider:
|
|
continue
|
|
if manager.find_plugin_skill(qualified_name) == registered_path:
|
|
manager.remove_plugin_skill(qualified_name)
|
|
registered.pop(qualified_name, None)
|
|
|
|
|
|
def discover_plugin_cli_commands() -> List[dict]:
|
|
"""CLI commands for the **active** memory plugin only. Imports just its ``cli.py``
|
|
(``register_cli(subparser)``), never the provider module, so it is safe during
|
|
argparse setup. At most one dict: name/help/description/setup_fn/handler_fn/plugin."""
|
|
active_provider = _get_active_memory_provider() if _MEMORY_PLUGINS_DIR.is_dir() else None
|
|
plugin_dir = find_provider_dir(active_provider) if active_provider else None
|
|
if not plugin_dir or not (plugin_dir / "cli.py").exists():
|
|
return []
|
|
|
|
module_name = _module_name(plugin_dir, active_provider) + ".cli"
|
|
try:
|
|
cli_mod = sys.modules.get(module_name)
|
|
if cli_mod is None:
|
|
if not _is_bundled(plugin_dir):
|
|
# cli.py imports as _hermes_user_memory.<name>.cli, usually before the
|
|
# provider is loaded: register parent packages so its relative imports
|
|
# resolve without executing the plugin's __init__.py (the shell has no
|
|
# __file__, so _load_provider_from_dir() still loads the real module).
|
|
_register_synthetic_package(_USER_NAMESPACE, [])
|
|
_register_synthetic_package(_module_name(plugin_dir, active_provider), [str(plugin_dir)])
|
|
spec = importlib.util.spec_from_file_location(module_name, str(plugin_dir / "cli.py"))
|
|
if not spec or not spec.loader:
|
|
return []
|
|
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 []
|
|
desc = _loader.read_plugin_description(plugin_dir)
|
|
return [{
|
|
"name": active_provider,
|
|
"help": desc or f"Manage {active_provider} memory plugin",
|
|
"description": desc or "",
|
|
"setup_fn": register_cli,
|
|
"handler_fn": getattr(cli_mod, f"{active_provider}_command", None) or getattr(cli_mod, "honcho_command", None),
|
|
"plugin": active_provider,
|
|
}]
|
|
except Exception as e:
|
|
logger.debug("Failed to scan CLI for memory plugin '%s': %s", active_provider, e)
|
|
return []
|