fix(plugins): only register CLI commands for the active memory provider
discover_plugin_cli_commands() now reads memory.provider from config.yaml and only loads CLI registration for the active provider. If no memory provider is set, no plugin CLI commands appear in the CLI. Only one memory provider can be active at a time — at most one set of plugin CLI commands is registered. Users who haven't configured honcho (or any memory provider) won't see 'hermes honcho' in their help output. Adds test for inactive provider returning empty results.
This commit is contained in:
Teknium
Teknium
parent
b074b0b13a
commit
0f813c422c
2 changed files with 120 additions and 62 deletions
|
|
@ -216,12 +216,33 @@ class _ProviderCollector:
|
|||
pass # CLI registration happens via discover_plugin_cli_commands()
|
||||
|
||||
|
||||
def discover_plugin_cli_commands() -> List[dict]:
|
||||
"""Scan memory plugin directories for CLI command registrations.
|
||||
def _get_active_memory_provider() -> Optional[str]:
|
||||
"""Read the active memory provider name from config.yaml.
|
||||
|
||||
Looks for a ``register_cli(subparser)`` function in each plugin's
|
||||
``cli.py``. Returns a list of dicts with keys:
|
||||
``name``, ``help``, ``description``, ``setup_fn``, ``handler_fn``.
|
||||
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 config.get("memory", {}).get("provider") or None
|
||||
except Exception:
|
||||
return 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
|
||||
|
|
@ -231,60 +252,66 @@ def discover_plugin_cli_commands() -> List[dict]:
|
|||
if not _MEMORY_PLUGINS_DIR.is_dir():
|
||||
return results
|
||||
|
||||
for child in sorted(_MEMORY_PLUGINS_DIR.iterdir()):
|
||||
if not child.is_dir() or child.name.startswith(("_", ".")):
|
||||
continue
|
||||
cli_file = child / "cli.py"
|
||||
if not cli_file.exists():
|
||||
continue
|
||||
active_provider = _get_active_memory_provider()
|
||||
if not active_provider:
|
||||
return results
|
||||
|
||||
module_name = f"plugins.memory.{child.name}.cli"
|
||||
try:
|
||||
# Import the CLI module (lightweight — no SDK needed)
|
||||
if module_name in sys.modules:
|
||||
cli_mod = sys.modules[module_name]
|
||||
else:
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
module_name, str(cli_file)
|
||||
)
|
||||
if not spec or not spec.loader:
|
||||
continue
|
||||
cli_mod = importlib.util.module_from_spec(spec)
|
||||
sys.modules[module_name] = cli_mod
|
||||
spec.loader.exec_module(cli_mod)
|
||||
# Only look at the active provider's directory
|
||||
plugin_dir = _MEMORY_PLUGINS_DIR / active_provider
|
||||
if not plugin_dir.is_dir():
|
||||
return results
|
||||
|
||||
register_cli = getattr(cli_mod, "register_cli", None)
|
||||
if not callable(register_cli):
|
||||
continue
|
||||
cli_file = plugin_dir / "cli.py"
|
||||
if not cli_file.exists():
|
||||
return results
|
||||
|
||||
# Read metadata from plugin.yaml if available
|
||||
help_text = f"Manage {child.name} memory plugin"
|
||||
description = ""
|
||||
yaml_file = child / "plugin.yaml"
|
||||
if yaml_file.exists():
|
||||
try:
|
||||
import yaml
|
||||
with open(yaml_file) as f:
|
||||
meta = yaml.safe_load(f) or {}
|
||||
desc = meta.get("description", "")
|
||||
if desc:
|
||||
help_text = desc
|
||||
description = desc
|
||||
except Exception:
|
||||
pass
|
||||
module_name = f"plugins.memory.{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:
|
||||
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)
|
||||
|
||||
handler_fn = getattr(cli_mod, "honcho_command", None) or \
|
||||
getattr(cli_mod, f"{child.name}_command", None)
|
||||
register_cli = getattr(cli_mod, "register_cli", None)
|
||||
if not callable(register_cli):
|
||||
return results
|
||||
|
||||
results.append({
|
||||
"name": child.name,
|
||||
"help": help_text,
|
||||
"description": description,
|
||||
"setup_fn": register_cli,
|
||||
"handler_fn": handler_fn,
|
||||
"plugin": child.name,
|
||||
})
|
||||
except Exception as e:
|
||||
logger.debug("Failed to scan CLI for memory plugin '%s': %s", child.name, e)
|
||||
# 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) 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
|
||||
|
|
|
|||
|
|
@ -80,8 +80,8 @@ class TestGetPluginCliCommands:
|
|||
|
||||
|
||||
class TestMemoryPluginCliDiscovery:
|
||||
def test_discovers_plugin_with_register_cli(self, tmp_path, monkeypatch):
|
||||
"""A memory plugin dir with cli.py containing register_cli is discovered."""
|
||||
def test_discovers_active_plugin_with_register_cli(self, tmp_path, monkeypatch):
|
||||
"""Only the active memory provider's CLI commands are discovered."""
|
||||
plugin_dir = tmp_path / "testplugin"
|
||||
plugin_dir.mkdir()
|
||||
(plugin_dir / "__init__.py").write_text("pass\n")
|
||||
|
|
@ -96,29 +96,58 @@ class TestMemoryPluginCliDiscovery:
|
|||
"name: testplugin\ndescription: A test plugin\n"
|
||||
)
|
||||
|
||||
# Patch _MEMORY_PLUGINS_DIR to our tmp dir
|
||||
# Also create a second plugin that should NOT be discovered
|
||||
other_dir = tmp_path / "otherplugin"
|
||||
other_dir.mkdir()
|
||||
(other_dir / "__init__.py").write_text("pass\n")
|
||||
(other_dir / "cli.py").write_text(
|
||||
"def register_cli(subparser):\n"
|
||||
" subparser.add_argument('--other')\n"
|
||||
)
|
||||
|
||||
import plugins.memory as pm
|
||||
original_dir = pm._MEMORY_PLUGINS_DIR
|
||||
|
||||
# Clear any cached module to force reimport
|
||||
mod_key = "plugins.memory.testplugin.cli"
|
||||
sys.modules.pop(mod_key, None)
|
||||
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", tmp_path)
|
||||
# Set testplugin as the active provider
|
||||
monkeypatch.setattr(pm, "_get_active_memory_provider", lambda: "testplugin")
|
||||
try:
|
||||
cmds = pm.discover_plugin_cli_commands()
|
||||
finally:
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", original_dir)
|
||||
sys.modules.pop(mod_key, None)
|
||||
|
||||
# Only testplugin should be discovered, not otherplugin
|
||||
assert len(cmds) == 1
|
||||
assert cmds[0]["name"] == "testplugin"
|
||||
assert cmds[0]["help"] == "A test plugin"
|
||||
assert callable(cmds[0]["setup_fn"])
|
||||
assert cmds[0]["handler_fn"].__name__ == "testplugin_command"
|
||||
|
||||
def test_returns_nothing_when_no_active_provider(self, tmp_path, monkeypatch):
|
||||
"""No commands when memory.provider is not set in config."""
|
||||
plugin_dir = tmp_path / "testplugin"
|
||||
plugin_dir.mkdir()
|
||||
(plugin_dir / "__init__.py").write_text("pass\n")
|
||||
(plugin_dir / "cli.py").write_text(
|
||||
"def register_cli(subparser):\n pass\n"
|
||||
)
|
||||
|
||||
import plugins.memory as pm
|
||||
original_dir = pm._MEMORY_PLUGINS_DIR
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", tmp_path)
|
||||
monkeypatch.setattr(pm, "_get_active_memory_provider", lambda: None)
|
||||
try:
|
||||
cmds = pm.discover_plugin_cli_commands()
|
||||
finally:
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", original_dir)
|
||||
|
||||
assert len(cmds) == 0
|
||||
|
||||
def test_skips_plugin_without_register_cli(self, tmp_path, monkeypatch):
|
||||
"""A memory plugin with cli.py but no register_cli is skipped."""
|
||||
"""An active plugin with cli.py but no register_cli returns nothing."""
|
||||
plugin_dir = tmp_path / "noplugin"
|
||||
plugin_dir.mkdir()
|
||||
(plugin_dir / "__init__.py").write_text("pass\n")
|
||||
|
|
@ -127,6 +156,7 @@ class TestMemoryPluginCliDiscovery:
|
|||
import plugins.memory as pm
|
||||
original_dir = pm._MEMORY_PLUGINS_DIR
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", tmp_path)
|
||||
monkeypatch.setattr(pm, "_get_active_memory_provider", lambda: "noplugin")
|
||||
try:
|
||||
cmds = pm.discover_plugin_cli_commands()
|
||||
finally:
|
||||
|
|
@ -136,7 +166,7 @@ class TestMemoryPluginCliDiscovery:
|
|||
assert len(cmds) == 0
|
||||
|
||||
def test_skips_plugin_without_cli_py(self, tmp_path, monkeypatch):
|
||||
"""A memory plugin dir without cli.py is skipped."""
|
||||
"""An active provider without cli.py returns nothing."""
|
||||
plugin_dir = tmp_path / "nocli"
|
||||
plugin_dir.mkdir()
|
||||
(plugin_dir / "__init__.py").write_text("pass\n")
|
||||
|
|
@ -144,6 +174,7 @@ class TestMemoryPluginCliDiscovery:
|
|||
import plugins.memory as pm
|
||||
original_dir = pm._MEMORY_PLUGINS_DIR
|
||||
monkeypatch.setattr(pm, "_MEMORY_PLUGINS_DIR", tmp_path)
|
||||
monkeypatch.setattr(pm, "_get_active_memory_provider", lambda: "nocli")
|
||||
try:
|
||||
cmds = pm.discover_plugin_cli_commands()
|
||||
finally:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue