Blame for plugins/memory/__init__.py

"""Memory provider plugin discovery.

Scans four sources for memory provider plugins:

1. Bundled providers: ``plugins/memory/<name>/`` (shipped with hermes-agent)
2. User-installed providers: ``$HERMES_HOME/plugins/<name>/``
3. Project-local providers: ``./.hermes/plugins/<name>/``, 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.<name>``, 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/<name>/)
    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/<name>/)
    # 3. Project-local providers (./.hermes/plugins/<name>/), 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 <provider>`` 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/<name>/``), user-installed
    (``$HERMES_HOME/plugins/<name>/``), 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.<name>.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