Merge #42076: nested category plugin discovery + alias-normalized enable/disable (#41066)

Merge #42076: nested category plugin discovery + alias-normalized enable/disable (#41066)

Lands the complete nested category plugin fix:
- Discovery in `hermes plugins list` (from @islam666's #41076, carried in this PR)
- Alias-normalized enable/disable mutation path so nested plugins can be toggled
- Fixes the #41076 base breakages (web_server 6-tuple unpack + stale test fixtures)

Co-authored work: discovery by @islam666 (#41076).
Closes #41066.
This commit is contained in:
kshitij 2026-06-08 05:47:27 -07:00 committed by GitHub
commit b99c6c4277
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 704 additions and 99 deletions

View file

@ -649,29 +649,62 @@ def _save_enabled_set(enabled: set) -> None:
save_config(config)
def _resolve_plugin_key(name: str) -> Optional[str]:
"""Resolve a user-supplied plugin identifier to its canonical registry key.
Accepts either the bare manifest name (``nemo_relay``), the directory
name, or the full path-derived key (``observability/nemo_relay``) and
returns the canonical key the loader gates on (``manifest.key`` or, for a
flat plugin, the bare name). Returns ``None`` when no plugin matches.
This is the single normalization point so ``hermes plugins enable`` /
``disable`` write the same key that ``PluginManager`` matches against
nested category plugins (e.g. ``observability/nemo_relay``) included.
"""
entries = _discover_all_plugins()
# 1. Exact match on canonical key or manifest name — always unambiguous.
for entry in entries:
# entry = (name, version, description, source, dir_path, key)
if name == entry[5] or name == entry[0]:
return entry[5]
# 2. Fall back to a bare leaf-name match (e.g. "nemo_relay" ->
# "observability/nemo_relay"), but only when it resolves to exactly one
# plugin so we never silently pick the wrong same-named nested plugin.
leaf_matches = [entry[5] for entry in entries if name == entry[5].split("/")[-1]]
if len(leaf_matches) == 1:
return leaf_matches[0]
return None
def cmd_enable(name: str) -> None:
"""Add a plugin to the enabled allow-list (and remove it from disabled)."""
from rich.console import Console
console = Console()
# Discover the plugin — check installed (user) AND bundled.
if not _plugin_exists(name):
# Discover the plugin — check installed (user) AND bundled, including
# nested category plugins — and normalize to its canonical registry key.
key = _resolve_plugin_key(name)
if key is None:
console.print(f"[red]Plugin '{name}' is not installed or bundled.[/red]")
sys.exit(1)
enabled = _get_enabled_set()
disabled = _get_disabled_set()
if name in enabled and name not in disabled:
console.print(f"[dim]Plugin '{name}' is already enabled.[/dim]")
if key in enabled and key not in disabled:
console.print(f"[dim]Plugin '{key}' is already enabled.[/dim]")
return
enabled.add(name)
disabled.discard(name)
enabled.add(key)
disabled.discard(key)
# Drop any legacy bare-name entry so the two don't drift out of sync.
bare = key.split("/")[-1]
if bare != key:
disabled.discard(bare)
_save_enabled_set(enabled)
_save_disabled_set(disabled)
console.print(
f"[green]✓[/green] Plugin [bold]{name}[/bold] enabled. "
f"[green]✓[/green] Plugin [bold]{key}[/bold] enabled. "
"Takes effect on next session."
)
@ -681,111 +714,129 @@ def cmd_disable(name: str) -> None:
from rich.console import Console
console = Console()
if not _plugin_exists(name):
key = _resolve_plugin_key(name)
if key is None:
console.print(f"[red]Plugin '{name}' is not installed or bundled.[/red]")
sys.exit(1)
enabled = _get_enabled_set()
disabled = _get_disabled_set()
if name not in enabled and name in disabled:
console.print(f"[dim]Plugin '{name}' is already disabled.[/dim]")
if key not in enabled and key in disabled:
console.print(f"[dim]Plugin '{key}' is already disabled.[/dim]")
return
enabled.discard(name)
disabled.add(name)
enabled.discard(key)
# Drop any legacy bare-name entry from the allow-list too, so a stale
# bare name can't keep a nested plugin loading after an explicit disable.
bare = key.split("/")[-1]
if bare != key:
enabled.discard(bare)
disabled.add(key)
_save_enabled_set(enabled)
_save_disabled_set(disabled)
console.print(
f"[yellow]\u2298[/yellow] Plugin [bold]{name}[/bold] disabled. "
f"[yellow]\u2298[/yellow] Plugin [bold]{key}[/bold] disabled. "
"Takes effect on next session."
)
def _plugin_exists(name: str) -> bool:
"""Return True if a plugin with *name* is installed (user) or bundled."""
# Installed: directory name or manifest name match in user plugins dir
user_dir = _plugins_dir()
if user_dir.is_dir():
if (user_dir / name).is_dir():
return True
for child in user_dir.iterdir():
if not child.is_dir():
continue
manifest = _read_manifest(child)
if manifest.get("name") == name:
return True
# Bundled: <repo>/plugins/<name>/ (or HERMES_BUNDLED_PLUGINS on Nix).
from hermes_cli.plugins import get_bundled_plugins_dir
repo_plugins = get_bundled_plugins_dir()
if repo_plugins.is_dir():
candidate = repo_plugins / name
if candidate.is_dir() and (
(candidate / "plugin.yaml").exists()
or (candidate / "plugin.yml").exists()
):
return True
return False
"""Return True if a plugin with *name* (bare name or key) exists."""
return _resolve_plugin_key(name) is not None
def _discover_all_plugins() -> list:
"""Return a list of (name, version, description, source, dir_path) for
every plugin the loader can see user + bundled + project.
def _read_manifest_info(d: Path, prefix: str):
"""Read a plugin.yaml manifest and return (name, version, description, key).
Matches the ordering/dedup of ``PluginManager.discover_and_load``:
bundled first, then user, then project; user overrides bundled on
name collision.
Returns None if no manifest file exists.
"""
manifest_file = d / "plugin.yaml"
if not manifest_file.exists():
manifest_file = d / "plugin.yml"
if not manifest_file.exists():
return None
try:
import yaml
except ImportError:
yaml = None
name = d.name
version = ""
description = ""
if yaml:
try:
with open(manifest_file, encoding="utf-8") as f:
manifest = yaml.safe_load(f) or {}
name = manifest.get("name", d.name)
version = manifest.get("version", "")
description = manifest.get("description", "")
except Exception:
pass
key = f"{prefix}/{d.name}" if prefix else name
return name, version, description, key
seen: dict = {} # name -> (name, version, description, source, path)
# Bundled (<repo>/plugins/<name>/), excluding memory/ and context_engine/
from hermes_cli.plugins import get_bundled_plugins_dir
repo_plugins = get_bundled_plugins_dir()
for base, source in ((repo_plugins, "bundled"), (_plugins_dir(), "user")):
if not base.is_dir():
def _scan_level(
base: Path,
source: str,
skip_names: set,
prefix: str,
depth: int,
seen: dict,
) -> None:
"""Recursive directory scan matching PluginManager._scan_directory_level.
Populates *seen* with key -> (name, version, description, source, dir, key).
"""
if not base.is_dir():
return
for d in sorted(base.iterdir()):
if not d.is_dir():
continue
for d in sorted(base.iterdir()):
if not d.is_dir():
continue
if source == "bundled" and d.name in {"memory", "context_engine"}:
continue
manifest_file = d / "plugin.yaml"
if not manifest_file.exists():
manifest_file = d / "plugin.yml"
if not manifest_file.exists():
continue
name = d.name
version = ""
description = ""
if yaml:
try:
with open(manifest_file, encoding="utf-8") as f:
manifest = yaml.safe_load(f) or {}
name = manifest.get("name", d.name)
version = manifest.get("version", "")
description = manifest.get("description", "")
except Exception:
pass
# User plugins override bundled on name collision.
if name in seen and source == "bundled":
if depth == 0 and skip_names and d.name in skip_names:
continue
info = _read_manifest_info(d, prefix)
if info is not None:
name, version, description, key = info
if key in seen and source == "bundled":
continue
src_label = source
if source == "user" and (d / ".git").exists():
src_label = "git"
seen[name] = (name, version, description, src_label, d)
seen[key] = (name, version, description, src_label, d, key)
continue
if depth >= 1:
continue
sub_prefix = f"{prefix}/{d.name}" if prefix else d.name
_scan_level(d, source, set(), sub_prefix, depth + 1, seen)
def _discover_all_plugins() -> list:
"""Return a list of (name, version, description, source, dir_path, key) for
every plugin the loader can see user + bundled + project.
Matches the ordering/dedup of ``PluginManager.discover_and_load``:
bundled first, then user, then project; user overrides bundled on
key collision.
"""
seen: dict = {} # key -> (name, version, description, source, path, key)
# Bundled (<repo>/plugins/<name>/), excluding memory/ and context_engine/
from hermes_cli.plugins import get_bundled_plugins_dir
repo_plugins = get_bundled_plugins_dir()
for base, source, skip in (
(repo_plugins, "bundled", {"memory", "context_engine"}),
(_plugins_dir(), "user", set()),
):
_scan_level(base, source, skip, "", 0, seen)
return list(seen.values())
def _plugin_status(name: str, enabled: set, disabled: set) -> str:
"""Return the user-facing activation state for a plugin name."""
if name in disabled:
def _plugin_status(name: str, enabled: set, disabled: set, key: str = "") -> str:
"""Return the user-facing activation state for a plugin name or key."""
if name in disabled or key in disabled:
return "disabled"
if name in enabled:
if name in enabled or key in enabled:
return "enabled"
return "not enabled"
@ -798,7 +849,7 @@ def _filter_plugin_entries(entries: list, args: Any, enabled: set, disabled: set
if getattr(args, "enabled", False):
filtered = [
entry for entry in filtered
if _plugin_status(entry[0], enabled, disabled) == "enabled"
if _plugin_status(entry[0], enabled, disabled, key=entry[5]) == "enabled"
]
return filtered
@ -823,19 +874,19 @@ def cmd_list(args: Any | None = None) -> None:
payload = [
{
"name": name,
"status": _plugin_status(name, enabled, disabled),
"status": _plugin_status(name, enabled, disabled, key=key),
"version": str(version),
"description": description,
"source": source,
}
for name, version, description, source, _dir in entries
for name, version, description, source, _dir, key in entries
]
print(json.dumps(payload, indent=2))
return
if getattr(args, "plain", False):
for name, version, _description, source, _dir in entries:
status = _plugin_status(name, enabled, disabled)
for name, version, _description, source, _dir, key in entries:
status = _plugin_status(name, enabled, disabled, key=key)
print(f"{status:12} {source:8} {str(version):8} {name}")
return
@ -850,8 +901,8 @@ def cmd_list(args: Any | None = None) -> None:
table.add_column("Description")
table.add_column("Source", style="dim")
for name, version, description, source, _dir in entries:
status_name = _plugin_status(name, enabled, disabled)
for name, version, description, source, _dir, key in entries:
status_name = _plugin_status(name, enabled, disabled, key=key)
if status_name == "disabled":
status = "[red]disabled[/red]"
elif status_name == "enabled":
@ -1051,14 +1102,14 @@ def cmd_toggle() -> None:
plugin_labels = []
plugin_selected = set()
for i, (name, _version, description, source, _d) in enumerate(entries):
for i, (name, _version, description, source, _d, key) in enumerate(entries):
label = f"{name} \u2014 {description}" if description else name
if source == "bundled":
label = f"{label} [bundled]"
plugin_names.append(name)
plugin_labels.append(label)
# Selected (enabled) when in enabled-set AND not in disabled-set
if name in enabled_set and name not in disabled_set:
if (name in enabled_set or key in enabled_set) and name not in disabled_set and key not in disabled_set:
plugin_selected.add(i)
# -- Provider categories --
@ -1641,7 +1692,7 @@ def _git_pull_plugin_dir(target: Path) -> tuple[bool, str]:
def dashboard_remove_user_plugin(name: str) -> dict[str, Any]:
"""Delete a plugin tree under ``~/.hermes/plugins/`` only."""
plugins_dir = _plugins_dir()
for n, _ver, _d, src, _path in _discover_all_plugins():
for n, _ver, _d, src, _path, _key in _discover_all_plugins():
if n == name and src == "bundled":
return {"ok": False, "error": "Bundled plugins cannot be removed from the dashboard."}

View file

@ -9575,10 +9575,16 @@ def _merged_plugins_hub() -> Dict[str, Any]:
plugins_root_resolved = (get_hermes_home() / "plugins").resolve()
rows: List[Dict[str, Any]] = []
for name, version, description, source, dir_str in _discover_all_plugins():
if name in disabled_set:
for name, version, description, source, dir_str, key in _discover_all_plugins():
# Both the path-derived key (nested category plugins) and the bare
# manifest name count for enabled/disabled state, matching the runtime
# loader's back-compat lookup.
aliases = {name}
if key:
aliases.add(key)
if aliases & disabled_set:
runtime_status = "disabled"
elif name in enabled_set:
elif aliases & enabled_set:
runtime_status = "enabled"
else:
runtime_status = "inactive"

View file

@ -0,0 +1,355 @@
"""Tests for the nested category plugin discovery fix (issue #41066).
Verifies that _discover_all_plugins() recurses into category directories
(up to 2 levels deep) and that _plugin_status() checks both manifest name
and path-derived key against the enabled/disabled sets.
"""
import json
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_plugin_dir(parent: Path, name: str, manifest: dict) -> Path:
"""Create a minimal plugin directory with a plugin.yaml."""
d = parent / name
d.mkdir(parents=True, exist_ok=True)
import yaml
(d / "plugin.yaml").write_text(yaml.dump(manifest), encoding="utf-8")
(d / "__init__.py").write_text("def register(ctx): pass\n", encoding="utf-8")
return d
def _make_category_plugin(
parent: Path, category: str, name: str, manifest: dict
) -> Path:
"""Create a category-namespaced plugin: <parent>/<category>/<name>/plugin.yaml."""
return _make_plugin_dir(parent / category, name, manifest)
# ---------------------------------------------------------------------------
# _read_manifest_info
# ---------------------------------------------------------------------------
class TestReadManifestInfo:
def test_flat_plugin(self, tmp_path):
from hermes_cli.plugins_cmd import _read_manifest_info
d = _make_plugin_dir(tmp_path, "my-plugin", {
"name": "my-plugin", "version": "1.0.0", "description": "test"
})
result = _read_manifest_info(d, "")
assert result is not None
name, version, description, key = result
assert name == "my-plugin"
assert version == "1.0.0"
assert description == "test"
assert key == "my-plugin" # flat: key == name
def test_category_plugin(self, tmp_path):
from hermes_cli.plugins_cmd import _read_manifest_info
d = _make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "2.0.0", "description": "search"
})
result = _read_manifest_info(d, "web")
assert result is not None
name, version, description, key = result
assert name == "web-tavily" # manifest name
assert key == "web/tavily" # path-derived key
def test_no_manifest(self, tmp_path):
from hermes_cli.plugins_cmd import _read_manifest_info
d = tmp_path / "empty-dir"
d.mkdir()
assert _read_manifest_info(d, "") is None
def test_yml_extension(self, tmp_path):
from hermes_cli.plugins_cmd import _read_manifest_info
d = tmp_path / "my-plugin"
d.mkdir()
import yaml
(d / "plugin.yml").write_text(yaml.dump({"name": "my-plugin"}), encoding="utf-8")
result = _read_manifest_info(d, "")
assert result is not None
assert result[0] == "my-plugin"
# ---------------------------------------------------------------------------
# _discover_all_plugins — recursive discovery
# ---------------------------------------------------------------------------
class TestDiscoverAllPlugins:
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_flat_plugins_still_discovered(self, mock_user_dir, mock_bundled_dir, tmp_path):
from hermes_cli.plugins_cmd import _discover_all_plugins
_make_plugin_dir(tmp_path, "disk-cleanup", {
"name": "disk-cleanup", "version": "1.0.0"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
entries = _discover_all_plugins()
keys = [e[5] for e in entries]
assert "disk-cleanup" in keys
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_category_plugins_discovered(self, mock_user_dir, mock_bundled_dir, tmp_path):
from hermes_cli.plugins_cmd import _discover_all_plugins
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0"
})
_make_category_plugin(tmp_path, "image_gen", "openai", {
"name": "image-gen-openai", "version": "2.0.0"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
entries = _discover_all_plugins()
keys = [e[5] for e in entries]
assert "web/tavily" in keys
assert "image_gen/openai" in keys
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_mixed_flat_and_category(self, mock_user_dir, mock_bundled_dir, tmp_path):
from hermes_cli.plugins_cmd import _discover_all_plugins
_make_plugin_dir(tmp_path, "disk-cleanup", {
"name": "disk-cleanup", "version": "1.0.0"
})
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0"
})
_make_category_plugin(tmp_path, "web", "exa", {
"name": "web-exa", "version": "1.0.0"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
entries = _discover_all_plugins()
keys = [e[5] for e in entries]
assert "disk-cleanup" in keys
assert "web/tavily" in keys
assert "web/exa" in keys
assert len(entries) == 3
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_depth_cap_at_two(self, mock_user_dir, mock_bundled_dir, tmp_path):
"""Plugins nested 3 levels deep should NOT be discovered."""
from hermes_cli.plugins_cmd import _discover_all_plugins
# 2 levels: should be found
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0"
})
# 3 levels: should NOT be found
deep = tmp_path / "a" / "b" / "c"
deep.mkdir(parents=True)
import yaml
(deep / "plugin.yaml").write_text(
yaml.dump({"name": "too-deep"}), encoding="utf-8"
)
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
entries = _discover_all_plugins()
keys = [e[5] for e in entries]
assert "web/tavily" in keys
assert "a/b/c" not in keys
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_tuple_has_six_elements(self, mock_user_dir, mock_bundled_dir, tmp_path):
from hermes_cli.plugins_cmd import _discover_all_plugins
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0", "description": "search"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
entries = _discover_all_plugins()
assert len(entries) == 1
entry = entries[0]
assert len(entry) == 6
name, version, description, source, dir_path, key = entry
assert name == "web-tavily"
assert key == "web/tavily"
assert source == "user"
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_user_overrides_bundled_on_key_collision(self, mock_user_dir, mock_bundled_dir, tmp_path):
"""User plugin with same key as bundled should win."""
from hermes_cli.plugins_cmd import _discover_all_plugins
# Simulate a bundled plugin
bundled_dir = tmp_path / "bundled"
bundled_dir.mkdir()
_make_plugin_dir(bundled_dir, "my-plugin", {
"name": "my-plugin", "version": "1.0.0"
})
# User plugin with same key
_make_plugin_dir(tmp_path, "my-plugin", {
"name": "my-plugin", "version": "2.0.0"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = bundled_dir
entries = _discover_all_plugins()
keys = [e[5] for e in entries]
assert keys.count("my-plugin") == 1
# User version should win
entry = [e for e in entries if e[5] == "my-plugin"][0]
assert entry[1] == "2.0.0"
# ---------------------------------------------------------------------------
# _plugin_status — key-aware status
# ---------------------------------------------------------------------------
class TestPluginStatus:
def test_name_in_enabled(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("my-plugin", {"my-plugin"}, set()) == "enabled"
def test_key_in_enabled(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("web-tavily", {"web/tavily"}, set(), key="web/tavily") == "enabled"
def test_name_in_disabled(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("my-plugin", set(), {"my-plugin"}) == "disabled"
def test_key_in_disabled(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("web-tavily", set(), {"web/tavily"}, key="web/tavily") == "disabled"
def test_neither_name_nor_key(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("unknown", {"other"}, set(), key="cat/unknown") == "not enabled"
def test_disabled_takes_precedence_over_enabled(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("my-plugin", {"my-plugin"}, {"my-plugin"}) == "disabled"
def test_key_disabled_takes_precedence(self):
from hermes_cli.plugins_cmd import _plugin_status
assert _plugin_status("web-tavily", {"web/tavily"}, {"web/tavily"}, key="web/tavily") == "disabled"
# ---------------------------------------------------------------------------
# Integration: _filter_plugin_entries with category plugins
# ---------------------------------------------------------------------------
class TestFilterPluginEntries:
def test_enabled_filter_uses_key(self):
from hermes_cli.plugins_cmd import _filter_plugin_entries
entries = [
("web-tavily", "1.0.0", "search", "user", Path("/tmp"), "web/tavily"),
("disk-cleanup", "1.0.0", "cleanup", "bundled", Path("/tmp"), "disk-cleanup"),
]
args = MagicMock()
args.no_bundled = False
args.user = False
args.enabled = True
result = _filter_plugin_entries(entries, args, {"web/tavily"}, set())
assert len(result) == 1
assert result[0][5] == "web/tavily"
def test_enabled_filter_by_name_still_works(self):
from hermes_cli.plugins_cmd import _filter_plugin_entries
entries = [
("disk-cleanup", "1.0.0", "cleanup", "bundled", Path("/tmp"), "disk-cleanup"),
]
args = MagicMock()
args.no_bundled = False
args.user = False
args.enabled = True
result = _filter_plugin_entries(entries, args, {"disk-cleanup"}, set())
assert len(result) == 1
# ---------------------------------------------------------------------------
# Integration: cmd_list JSON output includes category plugins
# ---------------------------------------------------------------------------
class TestCmdListJson:
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_json_output_includes_category_plugins(self, mock_user_dir, mock_bundled_dir, tmp_path, capsys):
from hermes_cli.plugins_cmd import cmd_list
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0", "description": "search"
})
_make_plugin_dir(tmp_path, "disk-cleanup", {
"name": "disk-cleanup", "version": "2.0.0", "description": "cleanup"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
args = MagicMock()
args.json = True
args.plain = False
args.no_bundled = False
args.user = False
args.enabled = False
cmd_list(args)
captured = capsys.readouterr()
payload = json.loads(captured.out)
names = [p["name"] for p in payload]
assert "web-tavily" in names
assert "disk-cleanup" in names
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_json_status_uses_key(self, mock_user_dir, mock_bundled_dir, tmp_path, capsys):
from hermes_cli.plugins_cmd import cmd_list
_make_category_plugin(tmp_path, "web", "tavily", {
"name": "web-tavily", "version": "1.0.0"
})
mock_user_dir.return_value = tmp_path
mock_bundled_dir.return_value = tmp_path / "nonexistent"
# Patch config to return web/tavily as enabled
with patch("hermes_cli.plugins_cmd._get_enabled_set", return_value={"web/tavily"}):
args = MagicMock()
args.json = True
args.plain = False
args.no_bundled = False
args.user = False
args.enabled = False
cmd_list(args)
captured = capsys.readouterr()
payload = json.loads(captured.out)
assert len(payload) == 1
assert payload[0]["status"] == "enabled"

View file

@ -0,0 +1,193 @@
"""Tests for nested/alias-normalized enable & disable flows.
Companion to test_plugins_cmd_category_discovery.py. That file covers the
*listing* side of nested category plugins (issue #41066). These tests cover
the *mutation* side: `hermes plugins enable/disable` must resolve a bare name
OR a full path-derived key (e.g. `observability/nemo_relay`) to the canonical
registry key and write THAT the same string PluginManager gates on so a
nested bundled plugin can actually be toggled.
"""
import sys # noqa: F401
from pathlib import Path
from unittest.mock import patch
import pytest
def _make_plugin_dir(parent: Path, name: str, manifest: dict) -> Path:
d = parent / name
d.mkdir(parents=True, exist_ok=True)
import yaml
(d / "plugin.yaml").write_text(yaml.dump(manifest), encoding="utf-8")
(d / "__init__.py").write_text("def register(ctx): pass\n", encoding="utf-8")
return d
def _make_category_plugin(parent: Path, category: str, name: str, manifest: dict) -> Path:
return _make_plugin_dir(parent / category, name, manifest)
@pytest.fixture
def nested_plugin_env(tmp_path):
"""A user-plugins dir containing one nested and one flat plugin, with the
bundled dir pointed at an empty path. Returns the tmp_path."""
_make_category_plugin(tmp_path, "observability", "nemo_relay", {
"name": "nemo_relay", "version": "1.0.0", "description": "relay obs"
})
_make_plugin_dir(tmp_path, "disk-cleanup", {
"name": "disk-cleanup", "version": "1.0.0"
})
return tmp_path
# ---------------------------------------------------------------------------
# _resolve_plugin_key
# ---------------------------------------------------------------------------
class TestResolvePluginKey:
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_full_key_resolves_to_itself(self, mock_user, mock_bundled, nested_plugin_env):
from hermes_cli.plugins_cmd import _resolve_plugin_key
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
assert _resolve_plugin_key("observability/nemo_relay") == "observability/nemo_relay"
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_bare_leaf_name_resolves_to_key(self, mock_user, mock_bundled, nested_plugin_env):
from hermes_cli.plugins_cmd import _resolve_plugin_key
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
# "nemo_relay" (bare) must normalize to the path-derived key.
assert _resolve_plugin_key("nemo_relay") == "observability/nemo_relay"
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_flat_plugin_resolves_to_name(self, mock_user, mock_bundled, nested_plugin_env):
from hermes_cli.plugins_cmd import _resolve_plugin_key
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
assert _resolve_plugin_key("disk-cleanup") == "disk-cleanup"
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_unknown_returns_none(self, mock_user, mock_bundled, nested_plugin_env):
from hermes_cli.plugins_cmd import _resolve_plugin_key
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
assert _resolve_plugin_key("does-not-exist") is None
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_ambiguous_leaf_name_returns_none(self, mock_user, mock_bundled, tmp_path):
"""Same leaf name under two categories must NOT silently pick one."""
from hermes_cli.plugins_cmd import _resolve_plugin_key
_make_category_plugin(tmp_path, "image_gen", "openai", {"name": "image-gen-openai"})
_make_category_plugin(tmp_path, "model-providers", "openai", {"name": "mp-openai"})
mock_user.return_value = tmp_path
mock_bundled.return_value = tmp_path / "nonexistent"
# Bare "openai" is ambiguous -> None; the full key still resolves.
assert _resolve_plugin_key("openai") is None
assert _resolve_plugin_key("image_gen/openai") == "image_gen/openai"
# ---------------------------------------------------------------------------
# cmd_enable / cmd_disable — write the canonical key
# ---------------------------------------------------------------------------
class TestEnableDisableNested:
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
@patch("hermes_cli.plugins_cmd._save_disabled_set")
@patch("hermes_cli.plugins_cmd._save_enabled_set")
@patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set())
@patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set())
def test_enable_bare_name_writes_key(
self, mock_en, mock_dis, mock_save_en, mock_save_dis,
mock_user, mock_bundled, nested_plugin_env,
):
from hermes_cli.plugins_cmd import cmd_enable
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
cmd_enable("nemo_relay") # bare name
saved = mock_save_en.call_args[0][0]
# The canonical key — NOT the bare name — must be persisted, because
# that is what PluginManager matches when deciding to load.
assert "observability/nemo_relay" in saved
assert "nemo_relay" not in saved or "observability/nemo_relay" in saved
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
@patch("hermes_cli.plugins_cmd._save_disabled_set")
@patch("hermes_cli.plugins_cmd._save_enabled_set")
@patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set())
@patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set())
def test_enable_full_key_writes_key(
self, mock_en, mock_dis, mock_save_en, mock_save_dis,
mock_user, mock_bundled, nested_plugin_env,
):
from hermes_cli.plugins_cmd import cmd_enable
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
cmd_enable("observability/nemo_relay")
saved = mock_save_en.call_args[0][0]
assert "observability/nemo_relay" in saved
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
@patch("hermes_cli.plugins_cmd._save_disabled_set")
@patch("hermes_cli.plugins_cmd._save_enabled_set")
@patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set())
@patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set())
def test_disable_bare_name_writes_key_and_clears_alias(
self, mock_en, mock_dis, mock_save_en, mock_save_dis,
mock_user, mock_bundled, nested_plugin_env,
):
from hermes_cli.plugins_cmd import cmd_disable
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
# Simulate an existing config where the plugin was enabled under the
# legacy bare name — disabling must clear that too, or the plugin would
# keep loading (PluginManager accepts the bare name as well).
mock_en.return_value = {"nemo_relay"}
cmd_disable("nemo_relay")
saved_dis = mock_save_dis.call_args[0][0]
saved_en = mock_save_en.call_args[0][0]
assert "observability/nemo_relay" in saved_dis
assert "nemo_relay" not in saved_en # stale bare alias dropped
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
def test_enable_unknown_plugin_exits(self, mock_user, mock_bundled, nested_plugin_env):
from hermes_cli.plugins_cmd import cmd_enable
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
with pytest.raises(SystemExit):
cmd_enable("does-not-exist")
@patch("hermes_cli.plugins.get_bundled_plugins_dir")
@patch("hermes_cli.plugins_cmd._plugins_dir")
@patch("hermes_cli.plugins_cmd._save_disabled_set")
@patch("hermes_cli.plugins_cmd._save_enabled_set")
@patch("hermes_cli.plugins_cmd._get_disabled_set", return_value=set())
@patch("hermes_cli.plugins_cmd._get_enabled_set", return_value=set())
def test_enable_flat_plugin_unchanged(
self, mock_en, mock_dis, mock_save_en, mock_save_dis,
mock_user, mock_bundled, nested_plugin_env,
):
"""Flat plugins keep writing their bare name (key == name) — no regression."""
from hermes_cli.plugins_cmd import cmd_enable
mock_user.return_value = nested_plugin_env
mock_bundled.return_value = nested_plugin_env / "nonexistent"
cmd_enable("disk-cleanup")
saved = mock_save_en.call_args[0][0]
assert "disk-cleanup" in saved

View file

@ -18,9 +18,9 @@ def _args(**kwargs):
def test_filter_plugin_entries_enabled_only():
entries = [
("disk-cleanup", "2.0.0", "Bundled", "bundled", None),
("web-search-plus", "2.2.0", "Search", "git", None),
("old-plugin", "1.0.0", "Old", "user", None),
("disk-cleanup", "2.0.0", "Bundled", "bundled", None, "disk-cleanup"),
("web-search-plus", "2.2.0", "Search", "git", None, "web-search-plus"),
("old-plugin", "1.0.0", "Old", "user", None, "old-plugin"),
]
filtered = plugins_cmd._filter_plugin_entries(
@ -35,9 +35,9 @@ def test_filter_plugin_entries_enabled_only():
def test_filter_plugin_entries_no_bundled():
entries = [
("disk-cleanup", "2.0.0", "Bundled", "bundled", None),
("drawthings-grpc", "0.3.0", "Draw Things", "user", None),
("web-search-plus", "2.2.0", "Search", "git", None),
("disk-cleanup", "2.0.0", "Bundled", "bundled", None, "disk-cleanup"),
("drawthings-grpc", "0.3.0", "Draw Things", "user", None, "drawthings-grpc"),
("web-search-plus", "2.2.0", "Search", "git", None, "web-search-plus"),
]
filtered = plugins_cmd._filter_plugin_entries(
@ -52,8 +52,8 @@ def test_filter_plugin_entries_no_bundled():
def test_cmd_list_plain_compact_output(monkeypatch, capsys):
entries = [
("disk-cleanup", "2.0.0", "Bundled", "bundled", None),
("web-search-plus", "2.2.0", "Search", "git", None),
("disk-cleanup", "2.0.0", "Bundled", "bundled", None, "disk-cleanup"),
("web-search-plus", "2.2.0", "Search", "git", None, "web-search-plus"),
]
monkeypatch.setattr(plugins_cmd, "_discover_all_plugins", lambda: entries)
monkeypatch.setattr(plugins_cmd, "_get_enabled_set", lambda: {"web-search-plus"})
@ -69,7 +69,7 @@ def test_cmd_list_plain_compact_output(monkeypatch, capsys):
def test_cmd_list_json_output(monkeypatch, capsys):
entries = [("web-search-plus", "2.2.0", "Search", "git", None)]
entries = [("web-search-plus", "2.2.0", "Search", "git", None, "web-search-plus")]
monkeypatch.setattr(plugins_cmd, "_discover_all_plugins", lambda: entries)
monkeypatch.setattr(plugins_cmd, "_get_enabled_set", lambda: {"web-search-plus"})
monkeypatch.setattr(plugins_cmd, "_get_disabled_set", lambda: set())