hermes-agent/plugins/memory/__init__.py

"""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 ast
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.
    A miss in the first 8 KB falls back to parsing, so a long module docstring cannot hide a provider."""
    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 or _defines_memory_provider(path)
    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 _defines_memory_provider(path: Path) -> bool:
    """Parse (never import) the package's top-level modules for what ``_load_provider_from_dir`` can load:
    a ``register_memory_provider(...)`` call or a ``MemoryProvider`` subclass. Unlike the discovery
    heuristic, a docstring or comment naming the contract does not count."""
    for module in sorted(path.glob("*.py")):
        try:
            tree = ast.parse(module.read_text(errors="replace", encoding="utf-8-sig"))
        except (OSError, SyntaxError, ValueError):
            continue
        bases = {"MemoryProvider"} | {a.asname for n in ast.walk(tree) if isinstance(n, ast.ImportFrom)
                                      for a in n.names if a.name == "MemoryProvider" and a.asname}
        for node in ast.walk(tree):
            if isinstance(node, ast.Call) and "register_memory_provider" in (
                    getattr(node.func, "attr", None), getattr(node.func, "id", None)):
                return True
            if isinstance(node, ast.ClassDef) and any(
                    (getattr(b, "attr", None) or getattr(b, "id", None)) in bases for b in node.bases):
                return True
    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` (or an update that could not install it) park a
        # user-installed provider in ``plugins.disabled``; the loader must honour it or "disabled" is a lie.
        logger.warning("%s", parked_provider_notice(name))
        return None

    def _load(_dir):
        if provider_dir:
            return _load_provider_from_dir(provider_dir, register_skills=register_skills)
        from hermes_cli.plugin_isolation import in_process_import_refusal
        refusal = in_process_import_refusal(f"pip-installed memory provider {name!r}")
        if refusal:
            logger.warning("%s", refusal)
            return None
        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
    from hermes_cli.plugin_isolation import ISOLATION_HOST, isolation_mode
    if isolation_mode() == ISOLATION_HOST and not _is_bundled(find_provider_dir(name) or Path("/")):
        return False  # third-party code runs in the plugin host; nothing to warm in this process
    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
    from hermes_cli.plugin_isolation import user_plugin_host
    host = None if _is_bundled(provider_dir) else user_plugin_host()
    if host is not None:
        return _ProviderCollector(name, register_skills=register_skills).collect_in_host(
            host, provider_dir, _module_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 collect_in_host(self, host, provider_dir: Path, module_name: str):
        """``collect`` for a provider whose code runs in the plugin host: the host captures the
        provider and forwards every other registration here. The discovery lock is NOT held across
        the host call — its registrations arrive on another thread and take the lock themselves."""
        from hermes_cli.plugins_ledger import _hook_source_of

        self._hook_source = _hook_source_of(self.name, SimpleNamespace(__file__=str(provider_dir / "__init__.py")))
        manager = self._plugin_context()._manager

        def drop_hooks() -> None:  # also before a reload into a restarted host: no stale hook proxies
            with manager._discovery_lock:
                manager._drop_fallback_hooks(self._hook_source)

        drop_hooks()
        self.provider = host.load_instance(
            provider_dir, module_name=module_name, capture="register_memory_provider",
            base_ref="agent.memory_provider:MemoryProvider", ctx=self, before_reload=drop_hooks)
        return self.provider

    def register_hook(self, hook_name, callback, *, hold_stream=False):
        context = self._plugin_context()
        with context._manager._discovery_lock:
            return context._manager._register_fallback_hook(
                context, self._hook_source, hook_name, callback, hold_stream=hold_stream)

    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 parked_provider_notice(name: str) -> Optional[str]:
    """The user-facing reason a configured provider is not loading because ``plugins.disabled`` parks
    it, naming the update that did so when there was one; None when it is not parked."""
    provider_dir = find_provider_dir(name)
    if provider_dir is None or not _explicitly_disabled(name, provider_dir):
        return None
    from hermes_cli.config import load_config
    from pm.plugins_state import RECORD
    record = cfg_get(load_config(), "plugins", RECORD)
    recorded = record.get(name) if isinstance(record, dict) else None
    why = (f" An update turned it off ({recorded.get('reason')}) and a later update retries it."
           if isinstance(recorded, dict) and recorded.get("reason") else "")
    return (f"⚠ Memory provider '{name}' is turned off: plugins.disabled lists it, so external memory is off "
            f"(built-in memory still works).{why} Turn it back on with `hermes plugins enable {name}`.")


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
    from pm.plugins_state import plugin_spellings

    return bool(plugin_spellings(provider_dir, name) & {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"
    if not _is_bundled(plugin_dir):
        from hermes_cli.plugin_isolation import in_process_import_refusal
        if in_process_import_refusal(f"memory provider {active_provider!r} CLI commands"):
            return []
    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),
            "plugin": active_provider,
        }]
    except Exception as e:
        logger.debug("Failed to scan CLI for memory plugin '%s': %s", active_provider, e)
        return []