From 8b70bf4c40b1f0d28fd5a9b9f3da4ad676bd68b9 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:14:51 -0700 Subject: [PATCH 1/4] feat(plugins): live catalog index with 6h cache and fallback Add fetch_live_catalog()/load_catalog_live() to hermes_cli.plugin_catalog: list plugin-catalog/*.yaml via the GitHub contents API (unauthenticated, 5s timeout), raw-fetch each entry, and cache under /cache/plugin-catalog/ with a 6h TTL. Any network failure falls back silently to the in-tree catalog. Also expose filter_entries() so callers can reuse search semantics on a live entry list. --- hermes_cli/plugin_catalog.py | 113 ++++++++++++++++++++++++++++++++++- 1 file changed, 111 insertions(+), 2 deletions(-) diff --git a/hermes_cli/plugin_catalog.py b/hermes_cli/plugin_catalog.py index 3de7c1ca9e6d..448f2c468451 100644 --- a/hermes_cli/plugin_catalog.py +++ b/hermes_cli/plugin_catalog.py @@ -23,6 +23,7 @@ from __future__ import annotations import logging import os import re +import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, List, Optional @@ -176,7 +177,11 @@ def load_catalog() -> List[PluginCatalogEntry]: Invalid entries are skipped with a logged warning; this function never raises for a malformed entry. """ - root = get_catalog_dir() + return _load_entries_from_dir(get_catalog_dir()) + + +def _load_entries_from_dir(root: Path) -> List[PluginCatalogEntry]: + """Parse all catalog entry files in *root* (skipping ``removed.yaml``).""" if not root.is_dir(): return [] entries: List[PluginCatalogEntry] = [] @@ -200,7 +205,17 @@ def get_catalog_entry(name: str) -> Optional[PluginCatalogEntry]: def search_catalog(query: str) -> List[PluginCatalogEntry]: """Case-insensitive substring search over name, description, and declared tools. An empty query returns the whole catalog.""" - entries = load_catalog() + return filter_entries(load_catalog(), query) + + +def filter_entries( + entries: List[PluginCatalogEntry], query: str +) -> List[PluginCatalogEntry]: + """Filter *entries* with :func:`search_catalog` semantics. + + Lets callers that already hold a (possibly live-fetched) entry list + apply the same matching rules without re-loading the catalog. + """ q = (query or "").strip().lower() if not q: return entries @@ -274,6 +289,100 @@ def find_removed(name_or_repo: str) -> Optional[RemovedEntry]: return None +# ─── Live index ────────────────────────────────────────────────────────────── + +# GitHub contents API for the in-repo catalog dir. Unauthenticated (60 req/hr +# rate limit) — fine for interactive use, and any failure falls back to the +# in-tree catalog silently. +_LIVE_INDEX_URL = ( + "https://api.github.com/repos/NousResearch/hermes-agent/contents/" + "plugin-catalog?ref=main" +) +_LIVE_TTL_SECONDS = 6 * 60 * 60 # 6h +_REQUEST_TIMEOUT = 5.0 + + +def _live_cache_dir() -> Path: + from hermes_constants import get_hermes_home + + return get_hermes_home() / "cache" / "plugin-catalog" + + +def fetch_live_catalog(*, force: bool = False) -> Optional[Path]: + """Refresh the catalog cache from the GitHub repo; return the cache dir. + + Lists ``plugin-catalog/*.yaml`` via the GitHub contents API, raw-fetches + each file, and stores them under ``/cache/plugin-catalog/`` + with a 6-hour TTL (repeat searches don't re-hit the API). Returns the + cache directory on success (or fresh cache), or ``None`` on ANY network + or parse failure — callers then fall back to the in-tree catalog. + """ + cache = _live_cache_dir() + marker = cache / ".fetched" + if not force and marker.is_file(): + try: + age = time.time() - marker.stat().st_mtime + except OSError: + age = _LIVE_TTL_SECONDS + 1 + if age < _LIVE_TTL_SECONDS: + return cache + + try: + import httpx + + resp = httpx.get( + _LIVE_INDEX_URL, + timeout=_REQUEST_TIMEOUT, + follow_redirects=True, + headers={"Accept": "application/vnd.github+json"}, + ) + resp.raise_for_status() + listing = resp.json() + if not isinstance(listing, list): + raise ValueError("unexpected contents-API payload") + + fetched: dict[str, str] = {} + for item in listing: + if not isinstance(item, dict): + continue + fname = str(item.get("name") or "") + url = str(item.get("download_url") or "") + if not fname.endswith(".yaml") or not url: + continue + file_resp = httpx.get( + url, timeout=_REQUEST_TIMEOUT, follow_redirects=True + ) + file_resp.raise_for_status() + fetched[fname] = file_resp.text + + cache.mkdir(parents=True, exist_ok=True) + # Replace stale cached entries wholesale so removed files disappear. + for old in cache.glob("*.yaml"): + if old.name not in fetched: + old.unlink(missing_ok=True) + for fname, text in fetched.items(): + (cache / fname).write_text(text, encoding="utf-8") + marker.touch() + return cache + except Exception as exc: + logger.debug("Plugin catalog: live index fetch failed: %s", exc) + return None + + +def load_catalog_live() -> List[PluginCatalogEntry]: + """Return catalog entries, preferring a live-fetched (or cached) index. + + Falls back silently to the in-tree catalog when the network is + unavailable or the fetch fails. + """ + cache = fetch_live_catalog() + if cache is not None and any( + p.name != "removed.yaml" for p in cache.glob("*.yaml") + ): + return _load_entries_from_dir(cache) + return load_catalog() + + # ─── Human summaries ───────────────────────────────────────────────────────── From 44624631bf0547a3836d2d6e84d85a62619ca246 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:14:51 -0700 Subject: [PATCH 2/4] feat(plugins): subprocess-isolated plugin validation for catalog CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New hermes_cli/plugin_validate.py — the checks behind 'hermes plugins validate ': manifest fields, strict requires_hermes spec parsing, config: shape, UPPER_SNAKE requires_env, a capability probe that imports the plugin and calls register(ctx) against a recording stub in a scratch subprocess (throwaway HERMES_HOME, 30s timeout) and diffs actual registrations against provides_* (undeclared = fail, unregistered = warn), plus built-in tool collision checks via the discovered tool registry. --- hermes_cli/plugin_validate.py | 434 ++++++++++++++++++++++++++++++++++ 1 file changed, 434 insertions(+) create mode 100644 hermes_cli/plugin_validate.py diff --git a/hermes_cli/plugin_validate.py b/hermes_cli/plugin_validate.py new file mode 100644 index 000000000000..74d0849a0514 --- /dev/null +++ b/hermes_cli/plugin_validate.py @@ -0,0 +1,434 @@ +"""``hermes plugins validate`` — admission checks for a plugin directory. + +This is the command the plugin-catalog admission CI (and the +``.github/actions/plugin-validate`` composite action) runs against a +candidate plugin. It performs static manifest checks plus a +subprocess-isolated capability probe: the plugin is imported and its +``register(ctx)`` called against a minimal recording stub context in a +scratch child process (with a throwaway ``HERMES_HOME``), so a crashing or +malicious plugin cannot take down the CLI, and the *actually registered* +tools/hooks/middleware are compared against the manifest's declared +``provides_*`` lists. +""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import sys +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +_UPPER_SNAKE_RE = re.compile(r"^[A-Z][A-Z0-9_]*$") +_CONFIG_TYPES = {"str", "bool", "int"} +_PROBE_TIMEOUT = 30 +_PROBE_SENTINEL = "HERMES_VALIDATE_JSON:" + + +@dataclass +class ValidationReport: + """Result of validating one plugin directory.""" + + checks: List[Tuple[str, bool, str]] = field(default_factory=list) + warnings: List[str] = field(default_factory=list) + + @property + def failures(self) -> List[str]: + return [detail or name for name, ok, detail in self.checks if not ok] + + @property + def ok(self) -> bool: + return all(ok for _name, ok, _detail in self.checks) + + @property + def exit_code(self) -> int: + return 0 if self.ok else 1 + + def add(self, name: str, ok: bool, detail: str = "") -> None: + self.checks.append((name, ok, detail)) + + def warn(self, message: str) -> None: + self.warnings.append(message) + + def to_dict(self) -> Dict[str, Any]: + return { + "ok": self.ok, + "checks": [ + {"name": name, "ok": ok, "detail": detail} + for name, ok, detail in self.checks + ], + "warnings": list(self.warnings), + } + + +# ─── Static checks ─────────────────────────────────────────────────────────── + + +def _requires_hermes_spec_valid(spec: str) -> bool: + """Strictly validate a ``requires_hermes`` spec. + + Unlike :func:`hermes_cli.plugins._version_satisfies` (permissive at load + time), validation REJECTS clauses whose version segment doesn't parse — + a typo'd spec should fail admission, not silently gate nothing. + """ + from hermes_cli.plugins import _VERSION_COMPARATOR_RE, _version_tuple + + for clause in spec.split(","): + clause = clause.strip() + if not clause: + continue + m = _VERSION_COMPARATOR_RE.match(clause) + target = m.group(2) if m else clause + if _version_tuple(target) is None: + return False + return True + + +def _check_manifest_fields(report: ValidationReport, manifest: dict) -> None: + missing = [ + f for f in ("name", "version", "description") if not manifest.get(f) + ] + if missing: + report.add( + "manifest fields", + False, + f"plugin.yaml missing required field(s): {', '.join(missing)}", + ) + else: + report.add("manifest fields", True, "name, version, description present") + + +def _check_requires_hermes(report: ValidationReport, manifest: dict) -> None: + spec = str(manifest.get("requires_hermes") or "").strip() + if not spec: + report.add("requires_hermes", True, "not declared") + return + if _requires_hermes_spec_valid(spec): + report.add("requires_hermes", True, f"spec {spec!r} parses") + else: + report.add( + "requires_hermes", + False, + f"requires_hermes spec {spec!r} does not parse " + "(expected e.g. \">=0.19\" or \">=0.19, <1.0\")", + ) + + +def _check_config_spec(report: ValidationReport, manifest: dict) -> None: + raw = manifest.get("config") + if raw in (None, [], {}): + report.add("config spec", True, "not declared") + return + problems: List[str] = [] + if not isinstance(raw, list): + problems.append("config: must be a list of mappings") + else: + for i, item in enumerate(raw): + if not isinstance(item, dict) or not item.get("key"): + problems.append(f"config[{i}]: must be a mapping with a 'key'") + continue + typ = item.get("type") + if typ is not None and str(typ) not in _CONFIG_TYPES: + problems.append( + f"config[{i}] ({item['key']}): type must be one of " + f"{'/'.join(sorted(_CONFIG_TYPES))}" + ) + secret = item.get("secret") + if secret is not None and not isinstance(secret, bool): + problems.append( + f"config[{i}] ({item['key']}): secret must be a boolean" + ) + if problems: + report.add("config spec", False, "; ".join(problems)) + else: + report.add("config spec", True, "shape valid") + + +def _check_requires_env(report: ValidationReport, manifest: dict) -> None: + raw = manifest.get("requires_env") or [] + problems: List[str] = [] + if not isinstance(raw, list): + problems.append("requires_env: must be a list") + raw = [] + for i, entry in enumerate(raw): + if isinstance(entry, str): + name = entry + elif isinstance(entry, dict): + name = str(entry.get("name") or "") + else: + problems.append(f"requires_env[{i}]: must be a string or mapping") + continue + if not _UPPER_SNAKE_RE.match(name): + problems.append( + f"requires_env[{i}]: {name!r} is not UPPER_SNAKE_CASE" + ) + if problems: + report.add("requires_env", False, "; ".join(problems)) + else: + report.add("requires_env", True, "all entries UPPER_SNAKE") + + +# ─── Capability probe (subprocess-isolated) ────────────────────────────────── + +# Self-contained harness run in a scratch child process. Imports the plugin +# module using the same file-location mechanics PluginManager uses, calls +# register() against a recording stub ctx, and prints a sentinel-prefixed +# JSON line of what was actually registered. Deliberately imports NOTHING +# from hermes so a hostile plugin only sees a bare interpreter. +_PROBE_SCRIPT = r""" +import importlib.util +import json +import sys + +plugin_dir = sys.argv[1] +sentinel = sys.argv[2] + +recorded = {"tools": [], "hooks": [], "middleware": [], "commands": []} + + +class RecordingContext: + plugin_config = {} + profile_name = "default" + + def register_tool(self, name, *args, **kwargs): + recorded["tools"].append(str(name)) + + def register_hook(self, hook_name, callback): + recorded["hooks"].append(str(hook_name)) + + def register_middleware(self, kind, callback): + recorded["middleware"].append(str(kind)) + + def register_command(self, name, *args, **kwargs): + recorded["commands"].append(str(name)) + + def register_cli_command(self, name, *args, **kwargs): + recorded["commands"].append(str(name)) + + def __getattr__(self, _name): + # Any other registration surface (platforms, providers, skills, + # context engines, ...) is accepted as a no-op — the probe only + # audits the declared-capability categories. + def _noop(*args, **kwargs): + return None + + return _noop + + +def emit(payload): + print(sentinel + json.dumps(payload)) + + +try: + spec = importlib.util.spec_from_file_location( + "hermes_validate_probe_plugin", + plugin_dir + "/__init__.py", + submodule_search_locations=[plugin_dir], + ) + module = importlib.util.module_from_spec(spec) + module.__path__ = [plugin_dir] + sys.modules[spec.name] = module + spec.loader.exec_module(module) +except Exception as exc: + emit({"error": "import failed: %s" % exc}) + sys.exit(0) + +register = getattr(module, "register", None) +if register is None: + emit({"error": "no register() function"}) + sys.exit(0) + +try: + register(RecordingContext()) +except Exception as exc: + emit({"error": "register() raised: %s" % exc}) + sys.exit(0) + +emit(recorded) +""" + + +def _run_capability_probe(plugin_dir: Path) -> Tuple[Optional[dict], str]: + """Run the recording probe in a scratch subprocess. + + Returns ``(recorded, error)`` — exactly one is meaningful: *recorded* + is the ``{tools, hooks, middleware, commands}`` dict on success, and + *error* is a human-readable failure description otherwise. + """ + with tempfile.TemporaryDirectory(prefix="hermes-validate-") as scratch: + env = dict(os.environ) + env["HERMES_HOME"] = scratch + try: + result = subprocess.run( + [ + sys.executable, + "-c", + _PROBE_SCRIPT, + str(plugin_dir), + _PROBE_SENTINEL, + ], + capture_output=True, + text=True, + timeout=_PROBE_TIMEOUT, + env=env, + ) + except subprocess.TimeoutExpired: + return None, f"capability probe timed out after {_PROBE_TIMEOUT}s" + + payload: Optional[dict] = None + for line in (result.stdout or "").splitlines(): + if line.startswith(_PROBE_SENTINEL): + try: + payload = json.loads(line[len(_PROBE_SENTINEL):]) + except json.JSONDecodeError: + payload = None + + if payload is None: + err = (result.stderr or "").strip() + return None, ( + "capability probe produced no result " + f"(exit {result.returncode})" + (f": {err}" if err else "") + ) + if "error" in payload: + return None, str(payload["error"]) + return payload, "" + + +def _declared_list(manifest: dict, key: str) -> List[str]: + raw = manifest.get(key) or [] + if not isinstance(raw, list): + return [] + return [str(item) for item in raw if isinstance(item, str)] + + +def _check_capabilities( + report: ValidationReport, manifest: dict, plugin_dir: Path +) -> Optional[dict]: + """Probe actual registrations and diff against declared capabilities. + + Returns the recorded dict (for the built-in collision check) or None + when the probe failed / was skipped. + """ + if not (plugin_dir / "__init__.py").is_file(): + report.warn( + "no __init__.py — capability probe skipped (manifest-only plugin)" + ) + report.add("capability probe", True, "skipped (no __init__.py)") + return None + + recorded, error = _run_capability_probe(plugin_dir) + if recorded is None: + report.add("capability probe", False, error) + return None + report.add("capability probe", True, "register() ran in isolation") + + for kind, manifest_key in ( + ("tools", "provides_tools"), + ("hooks", "provides_hooks"), + ("middleware", "provides_middleware"), + ): + declared = set(_declared_list(manifest, manifest_key)) + actual = set(recorded.get(kind) or []) + undeclared = sorted(actual - declared) + unregistered = sorted(declared - actual) + if undeclared: + report.add( + f"declared {kind}", + False, + f"undeclared {kind} registered (not in {manifest_key}): " + f"{', '.join(undeclared)}", + ) + else: + report.add(f"declared {kind}", True, "matches registrations") + if unregistered: + report.warn( + f"{manifest_key} declares {', '.join(unregistered)} " + f"but register() did not register them" + ) + return recorded + + +def _builtin_tool_names() -> List[str]: + """Return the built-in tool registry names (discovery-timing safe). + + ``tools.registry`` starts empty — built-in tool modules self-register on + import, so we must run ``discover_builtin_tools()`` first (idempotent; + see the AGENTS.md discover_plugins timing pitfall). + """ + try: + from tools.registry import discover_builtin_tools, registry + + discover_builtin_tools() + return list(registry.get_all_tool_names()) + except Exception: + return [] + + +def _check_builtin_collisions( + report: ValidationReport, manifest: dict, recorded: Optional[dict] +) -> None: + candidate_tools = set(_declared_list(manifest, "provides_tools")) + if recorded: + candidate_tools.update(recorded.get("tools") or []) + if not candidate_tools: + report.add("built-in tool collisions", True, "no tools to check") + return + builtin = set(_builtin_tool_names()) + collisions = sorted(candidate_tools & builtin) + if collisions: + report.add( + "built-in tool collisions", + False, + "tool name(s) collide with built-in tools: " + f"{', '.join(collisions)}", + ) + else: + report.add("built-in tool collisions", True, "no collisions") + + +# ─── Entry point ───────────────────────────────────────────────────────────── + + +def validate_plugin_dir(plugin_dir: Path) -> ValidationReport: + """Run every admission check against *plugin_dir* and return the report.""" + report = ValidationReport() + plugin_dir = Path(plugin_dir) + + if not plugin_dir.is_dir(): + report.add( + "plugin directory", False, f"{plugin_dir} is not a directory" + ) + return report + + manifest_file = plugin_dir / "plugin.yaml" + if not manifest_file.is_file(): + manifest_file = plugin_dir / "plugin.yml" + if not manifest_file.is_file(): + report.add("manifest", False, "no plugin.yaml in the plugin directory") + return report + + import yaml + + try: + manifest = yaml.safe_load( + manifest_file.read_text(encoding="utf-8") + ) + except Exception as exc: + report.add("manifest", False, f"plugin.yaml failed to parse: {exc}") + return report + if not isinstance(manifest, dict): + report.add("manifest", False, "plugin.yaml must be a mapping") + return report + report.add("manifest", True, "plugin.yaml parses") + + _check_manifest_fields(report, manifest) + _check_requires_hermes(report, manifest) + _check_config_spec(report, manifest) + _check_requires_env(report, manifest) + recorded = _check_capabilities(report, manifest, plugin_dir) + _check_builtin_collisions(report, manifest, recorded) + return report From f1b30414d5fd1ef49ef70d368d57da1e6b87abcb Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:14:51 -0700 Subject: [PATCH 3/4] =?UTF-8?q?feat(plugins):=20catalog=20CLI=20surface=20?= =?UTF-8?q?=E2=80=94=20search/browse/info/install/update/doctor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - install: catalog names resolve to the pinned SHA (ref= checkout), print tier + capability summary before the enable prompt, and write a .hermes-catalog.json provenance sidecar; raw git URLs get a 'custom (unreviewed) source' banner; --allow-removed loudly bypasses the removed blocklist (skip_removed_check=True) - update: sidecar installs compare against the current catalog pin and force-reinstall at the new SHA (enabled state preserved); plain git installs keep the git-pull flow - list: catalog:@ annotation + red 'REMOVED from catalog' lines for blocklisted installs (table and --json) - search/browse/info: live-index catalog tables and full entry detail with removed-list warnings - validate: human ✓/✗ output, --json for CI, exit 0/1 - doctor: per-plugin manifest/enabled/load-error/env/requires_hermes/ provenance/pin/removed diagnosis, compact table or single-name detail --- hermes_cli/plugins_cmd.py | 607 +++++++++++++++++++++++++++++- hermes_cli/subcommands/plugins.py | 45 +++ 2 files changed, 645 insertions(+), 7 deletions(-) diff --git a/hermes_cli/plugins_cmd.py b/hermes_cli/plugins_cmd.py index 731f35c6efa4..7ec4f9e1f4d1 100644 --- a/hermes_cli/plugins_cmd.py +++ b/hermes_cli/plugins_cmd.py @@ -613,12 +613,98 @@ def _install_plugin_core( return target, installed_manifest, installed_name +# --------------------------------------------------------------------------- +# Catalog integration helpers +# --------------------------------------------------------------------------- + +_CATALOG_SIDECAR = ".hermes-catalog.json" + + +def _looks_like_catalog_name(identifier: str) -> bool: + """True when *identifier* could be a catalog entry name (not a URL/shorthand).""" + if not identifier or "/" in identifier or "\\" in identifier: + return False + if identifier.startswith(("https://", "http://", "git@", "ssh://", "file://")): + return False + from hermes_cli.plugin_catalog import _NAME_RE + + return bool(_NAME_RE.match(identifier)) + + +def _get_live_catalog_entry(name: str): + """Look up *name* in the live-refreshed catalog (falls back in-tree).""" + from hermes_cli.plugin_catalog import load_catalog_live + + for entry in load_catalog_live(): + if entry.name == name: + return entry + return None + + +def _catalog_install_identifier(entry) -> str: + """Build the ``_install_plugin_core`` identifier for a catalog entry. + + Uses the explicit ``#subdir`` fragment form understood by + :func:`_resolve_git_url` when the entry lives in a repo subdirectory. + """ + if entry.subdir: + return f"{entry.repo}#{entry.subdir}" + return entry.repo + + +def _write_catalog_sidecar(target: Path, entry) -> None: + """Record catalog provenance in ``.hermes-catalog.json`` inside *target*. + + The sidecar is how ``update``/``list``/``doctor`` know the plugin came + from the catalog (and at which pin). + """ + import datetime + + sidecar = { + "catalog_name": entry.name, + "repo": entry.repo, + "sha": entry.sha, + "installed_at": datetime.datetime.now(datetime.timezone.utc) + .isoformat(timespec="seconds") + .replace("+00:00", "Z"), + "tier": entry.tier, + } + try: + (target / _CATALOG_SIDECAR).write_text( + json.dumps(sidecar, indent=2) + "\n", encoding="utf-8" + ) + except OSError as exc: + logger.warning("Failed to write catalog sidecar in %s: %s", target, exc) + + +def _read_catalog_sidecar(plugin_dir: Path) -> Optional[dict]: + """Return the parsed catalog provenance sidecar, or None.""" + path = plugin_dir / _CATALOG_SIDECAR + if not path.is_file(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + # Unreadable / corrupt sidecar — treat as a non-catalog install. + return None + return data if isinstance(data, dict) else None + + def cmd_install( identifier: str, force: bool = False, enable: Optional[bool] = None, + allow_removed: bool = False, ) -> None: - """Install a plugin from a Git URL or owner/repo shorthand. + """Install a plugin from the catalog, a Git URL, or owner/repo shorthand. + + When *identifier* matches a catalog entry name (and is not a URL or + ``owner/repo`` shorthand), the install resolves to the entry's pinned + commit SHA and records catalog provenance in a ``.hermes-catalog.json`` + sidecar. Raw git URLs keep the direct flow but are flagged as custom + (unreviewed) sources. + + ``allow_removed=True`` bypasses the removed-blocklist check (loudly). After install, prompt "Enable now? [y/N]" unless *enable* is provided (True = auto-enable without prompting, False = install disabled). @@ -627,6 +713,51 @@ def cmd_install( console = Console() + entry = None + if _looks_like_catalog_name(identifier): + from hermes_cli.plugin_catalog import ( + entry_capability_summary, + find_removed, + ) + + entry = _get_live_catalog_entry(identifier) + if entry is None: + console.print( + f"[red]Error:[/red] '{identifier}' is not in the Hermes " + "plugin catalog and is not a Git URL or owner/repo " + "shorthand.\n" + "Browse available entries with `hermes plugins search`." + ) + sys.exit(1) + if not allow_removed: + removed = find_removed(entry.name) or find_removed(entry.repo) + if removed is not None: + try: + _raise_removed(removed) + except PluginOperationError as e: + console.print(f"[red]Error:[/red] {e}") + sys.exit(1) + console.print( + f"[bold]{entry.name}[/bold] " + f"[cyan]\\[{entry.tier}][/cyan] " + f"[dim]pinned @ {entry.sha[:8]}[/dim]" + ) + console.print(entry_capability_summary(entry)) + identifier = _catalog_install_identifier(entry) + else: + console.print( + "[yellow]Warning:[/yellow] custom (unreviewed) source — " + "not from the Hermes catalog." + ) + + if allow_removed: + console.print( + "[bold red]WARNING:[/bold red] [red]--allow-removed set — " + "skipping the removed-plugin blocklist check. This plugin may " + "have been removed from the catalog for security reasons. " + "Proceed at your own risk.[/red]" + ) + try: git_url, _subdir = _resolve_git_url(identifier) except ValueError as e: @@ -648,11 +779,16 @@ def cmd_install( target, installed_manifest, installed_name = _install_plugin_core( identifier, force=force, + ref=entry.sha if entry is not None else None, + skip_removed_check=allow_removed, ) except PluginOperationError as e: console.print(f"[red]Error:[/red] {e}") sys.exit(1) + if entry is not None: + _write_catalog_sidecar(target, entry) + if not (target / "plugin.yaml").exists() and not (target / "plugin.yml").exists() and not ( target / "__init__.py" ).exists(): @@ -700,7 +836,13 @@ def cmd_install( def cmd_update(name: str) -> None: - """Update an installed plugin by pulling latest from its git remote.""" + """Update an installed plugin. + + Catalog installs (``.hermes-catalog.json`` sidecar present) are compared + against the current catalog pin: if the pinned SHA changed, the plugin is + force-reinstalled at the new pin (enabled state preserved). Plain git + installs keep the existing ``git pull`` behavior. + """ from rich.console import Console console = Console() @@ -712,6 +854,11 @@ def cmd_update(name: str) -> None: console.print(f"[red]Error:[/red] {e}") sys.exit(1) + sidecar = _read_catalog_sidecar(target) + if sidecar is not None and sidecar.get("catalog_name"): + _update_catalog_plugin(name, target, sidecar, console) + return + if not (target / ".git").exists(): console.print( f"[red]Error:[/red] Plugin '{name}' was not installed from git " @@ -739,6 +886,55 @@ def cmd_update(name: str) -> None: console.print(f"[dim]{out}[/dim]") +def _update_catalog_plugin(name: str, target: Path, sidecar: dict, console) -> None: + """Re-pin a catalog-installed plugin to the current catalog SHA.""" + catalog_name = str(sidecar.get("catalog_name") or name) + entry = _get_live_catalog_entry(catalog_name) + if entry is None: + console.print( + f"[red]Error:[/red] Plugin '{catalog_name}' is no longer in the " + "catalog — it may have been removed. Check " + "`hermes plugins doctor` and the removed blocklist." + ) + sys.exit(1) + + installed_sha = str(sidecar.get("sha") or "").strip().lower() + if installed_sha == entry.sha: + console.print( + f"[green]✓[/green] Plugin [bold]{catalog_name}[/bold] is " + f"already at catalog pin ({entry.sha[:8]})." + ) + return + + console.print( + f"[dim]Updating {catalog_name} to catalog pin:[/dim] " + f"{installed_sha[:8] or '(unknown)'} → {entry.sha[:8]}" + ) + + # Preserve enabled state across the force reinstall. + was_enabled = _get_enabled_set() + + try: + new_target, _manifest, _installed_name = _install_plugin_core( + _catalog_install_identifier(entry), + force=True, + ref=entry.sha, + ) + except PluginOperationError as e: + console.print(f"[red]Error:[/red] {e}") + sys.exit(1) + + _write_catalog_sidecar(new_target, entry) + # Restore the pre-update enabled/disabled state verbatim (the reinstall + # itself never touches it, but be explicit in case core ever does). + _save_enabled_set(was_enabled) + + console.print( + f"[green]✓[/green] Plugin [bold]{catalog_name}[/bold] updated to " + f"{entry.sha[:8]}." + ) + + def cmd_remove(name: str) -> None: """Remove an installed plugin by name.""" from rich.console import Console @@ -1188,6 +1384,44 @@ def _filter_plugin_entries(entries: list, args: Any, enabled: set, disabled: set return filtered +def _catalog_annotation(dir_path) -> Optional[str]: + """Return ``catalog:@`` for a catalog install, else None.""" + if not dir_path: + return None + try: + sidecar = _read_catalog_sidecar(Path(dir_path)) + except Exception: + return None + if not sidecar or not sidecar.get("catalog_name"): + return None + tier = str(sidecar.get("tier") or "community") + sha = str(sidecar.get("sha") or "") + return f"catalog:{tier}@{sha[:8]}" + + +def _removed_annotation(name: str, dir_path) -> Optional[str]: + """Return the removed-blocklist reason when *name* matches, else None.""" + try: + from hermes_cli.plugin_catalog import find_removed + except Exception: + return None + candidates = [name] + if dir_path: + try: + sidecar = _read_catalog_sidecar(Path(dir_path)) + except Exception: + sidecar = None + if sidecar: + candidates.extend( + str(v) for v in (sidecar.get("catalog_name"), sidecar.get("repo")) if v + ) + for candidate in candidates: + removed = find_removed(candidate) + if removed is not None: + return removed.reason or "no reason recorded" + return None + + def cmd_list(args: Any | None = None) -> None: """List all plugins (bundled + user) with enabled/disabled state.""" from rich.console import Console @@ -1205,16 +1439,22 @@ def cmd_list(args: Any | None = None) -> None: entries = _filter_plugin_entries(entries, args, enabled, disabled) if getattr(args, "json", False): - payload = [ - { + payload = [] + for name, version, description, source, _dir, key in entries: + row = { "name": name, "status": _plugin_status(name, enabled, disabled, key=key), "version": str(version), "description": description, "source": source, } - for name, version, description, source, _dir, key in entries - ] + catalog = _catalog_annotation(_dir) + if catalog: + row["catalog"] = catalog + removed_reason = _removed_annotation(name, _dir) + if removed_reason is not None: + row["removed"] = removed_reason + payload.append(row) print(json.dumps(payload, indent=2)) return @@ -1235,6 +1475,7 @@ def cmd_list(args: Any | None = None) -> None: table.add_column("Description") table.add_column("Source", style="dim") + removed_lines: list[str] = [] for name, version, description, source, _dir, key in entries: status_name = _plugin_status(name, enabled, disabled, key=key) if status_name == "disabled": @@ -1243,10 +1484,20 @@ def cmd_list(args: Any | None = None) -> None: status = "[green]enabled[/green]" else: status = "[yellow]not enabled[/yellow]" - table.add_row(name, status, str(version), description, source) + catalog = _catalog_annotation(_dir) + source_label = f"{source} [cyan]{catalog}[/cyan]" if catalog else source + table.add_row(name, status, str(version), description, source_label) + removed_reason = _removed_annotation(name, _dir) + if removed_reason is not None: + removed_lines.append( + f"[red bold]✗ {name} — REMOVED from catalog: " + f"{removed_reason}[/red bold]" + ) console.print() console.print(table) + for line in removed_lines: + console.print(line) console.print() console.print("[dim]Compact view:[/dim] hermes plugins list --plain --no-bundled") console.print("[dim]Interactive toggle:[/dim] hermes plugins") @@ -1254,6 +1505,337 @@ def cmd_list(args: Any | None = None) -> None: console.print("[dim]Plugins are opt-in by default — only 'enabled' plugins load.[/dim]") +# --------------------------------------------------------------------------- +# Catalog commands — search / browse / info / validate / doctor +# --------------------------------------------------------------------------- + + +def _entry_capability_counts(entry) -> str: + """Compact capability summary like ``2 tools, 1 hook`` for table rows.""" + caps = entry.capabilities + parts: list[str] = [] + for count, singular in ( + (len(caps.provides_tools), "tool"), + (len(caps.provides_hooks), "hook"), + (len(caps.provides_middleware), "middleware"), + ): + if count: + plural = "" if count == 1 or singular == "middleware" else "s" + parts.append(f"{count} {singular}{plural}") + if caps.requires_env: + parts.append(f"{len(caps.requires_env)} env") + return ", ".join(parts) or "—" + + +def _render_catalog_entries(entries, console) -> None: + """Render catalog entries as the shared search/browse Rich table.""" + from rich.table import Table + + table = Table(title="Hermes Plugin Catalog", show_lines=False) + table.add_column("Name", style="bold") + table.add_column("Tier") + table.add_column("Description") + table.add_column("Pinned", style="dim") + table.add_column("Capabilities", style="dim") + + for entry in entries: + tier = ( + "[cyan]official[/cyan]" + if entry.tier == "official" + else "[magenta]community[/magenta]" + ) + description = entry.description + if len(description) > 60: + description = description[:57] + "..." + table.add_row( + entry.name, + tier, + description, + entry.sha[:8], + _entry_capability_counts(entry), + ) + + console.print() + console.print(table) + console.print() + console.print("[dim]Details:[/dim] hermes plugins info ") + console.print("[dim]Install:[/dim] hermes plugins install ") + + +def cmd_search(query: str = "") -> None: + """Search the plugin catalog (live index when reachable).""" + from rich.console import Console + + from hermes_cli.plugin_catalog import filter_entries, load_catalog_live + + console = Console() + entries = filter_entries(load_catalog_live(), query) + if not entries: + if query: + console.print( + f"[dim]No catalog entries match '{query}'. " + "Browse everything with `hermes plugins browse`.[/dim]" + ) + else: + console.print("[dim]No catalog entries available.[/dim]") + return + _render_catalog_entries(entries, console) + + +def cmd_browse() -> None: + """List every plugin catalog entry.""" + cmd_search("") + + +def cmd_info(name: str) -> None: + """Show the full catalog entry for *name*.""" + from rich.console import Console + + from hermes_cli.plugin_catalog import find_removed + + console = Console() + entry = _get_live_catalog_entry(name) + if entry is None: + console.print( + f"[red]Error:[/red] '{name}' is not in the plugin catalog. " + "Browse entries with `hermes plugins search`." + ) + sys.exit(1) + + caps = entry.capabilities + console.print() + console.print(f"[bold]{entry.name}[/bold] [cyan]\\[{entry.tier}][/cyan]") + if entry.description: + console.print(entry.description) + console.print() + console.print(f"[dim]Repo:[/dim] {entry.repo}") + if entry.subdir: + console.print(f"[dim]Subdir:[/dim] {entry.subdir}") + console.print(f"[dim]Pinned SHA:[/dim] {entry.sha}") + console.print(f"[dim]Maintainer:[/dim] {entry.maintainer}") + if entry.requires_hermes: + console.print(f"[dim]Requires:[/dim] hermes {entry.requires_hermes}") + if entry.platforms: + console.print(f"[dim]Platforms:[/dim] {', '.join(entry.platforms)}") + if entry.docs_url: + console.print(f"[dim]Docs:[/dim] {entry.docs_url}") + console.print() + console.print(f"[dim]Tools:[/dim] {', '.join(caps.provides_tools) or '(none)'}") + console.print(f"[dim]Hooks:[/dim] {', '.join(caps.provides_hooks) or '(none)'}") + console.print(f"[dim]Middleware:[/dim] {', '.join(caps.provides_middleware) or '(none)'}") + console.print(f"[dim]Env vars:[/dim] {', '.join(caps.requires_env) or '(none)'}") + console.print() + + removed = find_removed(entry.name) or find_removed(entry.repo) + if removed is not None: + detail = removed.reason or "no reason recorded" + if removed.date: + detail += f" (removed {removed.date})" + console.print( + f"[red bold]✗ REMOVED from catalog: {detail}[/red bold]" + ) + console.print() + + console.print(f"[dim]Install:[/dim] hermes plugins install {entry.name}") + console.print() + + +def cmd_validate(path: str, as_json: bool = False) -> None: + """Validate a plugin directory for catalog admission. Exits 0/1.""" + from rich.console import Console + + from hermes_cli.plugin_validate import validate_plugin_dir + + report = validate_plugin_dir(Path(path)) + + if as_json: + print(json.dumps(report.to_dict(), indent=2)) + sys.exit(report.exit_code) + + console = Console() + console.print() + for name, ok, detail in report.checks: + mark = "[green]✓[/green]" if ok else "[red]✗[/red]" + line = f"{mark} {name}" + if detail: + line += f" [dim]— {detail}[/dim]" + console.print(line) + for warning in report.warnings: + console.print(f"[yellow]⚠ {warning}[/yellow]") + console.print() + if report.ok: + console.print("[green bold]Validation passed.[/green bold]") + else: + console.print("[red bold]Validation failed.[/red bold]") + sys.exit(report.exit_code) + + +def _runtime_load_errors() -> dict[str, str]: + """Return ``{plugin_key: error}`` from a fresh PluginManager scan. + + Uses the idempotent ``discover_plugins()`` path so we don't need a full + agent boot; failures are non-fatal (doctor still reports what it can). + """ + try: + from hermes_cli.plugins import discover_plugins, get_plugin_manager + + discover_plugins() + manager = get_plugin_manager() + return { + key: loaded.error + for key, loaded in manager._plugins.items() + if loaded.error + } + except Exception as exc: + logger.debug("doctor: runtime plugin scan failed: %s", exc) + return {} + + +def _doctor_plugin_report(name: str, dir_path: Path, key: str, + enabled: set, disabled: set, + load_errors: dict[str, str]) -> dict: + """Collect doctor facts for one installed plugin directory.""" + from hermes_cli.plugins import _running_hermes_version, _version_satisfies + + manifest = _read_manifest(dir_path) + facts: dict[str, Any] = { + "name": name, + "manifest_ok": bool(manifest.get("name")), + "status": _plugin_status(name, enabled, disabled, key=key), + "load_error": load_errors.get(key) or load_errors.get(name) or "", + "missing_env": _missing_requires_env_names(manifest), + "requires_hermes": "", + "catalog": "", + "pin": "", + "removed": "", + } + + spec = str(manifest.get("requires_hermes") or "").strip() + if spec: + current = _running_hermes_version() + if _version_satisfies(spec, current): + facts["requires_hermes"] = f"{spec} ✓" + else: + facts["requires_hermes"] = f"{spec} ✗ (running {current})" + + sidecar = _read_catalog_sidecar(dir_path) + if sidecar and sidecar.get("catalog_name"): + tier = str(sidecar.get("tier") or "community") + sha = str(sidecar.get("sha") or "") + facts["catalog"] = f"catalog:{tier}@{sha[:8]}" + entry = _get_live_catalog_entry(str(sidecar["catalog_name"])) + if entry is None: + facts["pin"] = "entry gone from catalog" + elif entry.sha != sha: + facts["pin"] = ( + f"behind catalog pin ({sha[:8]} → {entry.sha[:8]}) — " + "run `hermes plugins update`" + ) + else: + facts["pin"] = "at catalog pin" + + removed_reason = _removed_annotation(name, dir_path) + if removed_reason is not None: + facts["removed"] = removed_reason + return facts + + +def cmd_doctor(name: Optional[str] = None) -> None: + """Diagnose installed plugins (or a single one when *name* given).""" + from rich.console import Console + from rich.table import Table + + console = Console() + plugins_dir = _plugins_dir() + installed = [ + (d.name, d) for d in sorted(plugins_dir.iterdir()) if d.is_dir() + ] + if name is not None: + installed = [(n, d) for n, d in installed if n == name] + if not installed: + console.print( + f"[red]Error:[/red] Plugin '{name}' not found in {plugins_dir}." + ) + sys.exit(1) + + if not installed: + console.print("[dim]No plugins installed under[/dim] " + f"{plugins_dir}") + return + + enabled = _get_enabled_set() + disabled = _get_disabled_set() + load_errors = _runtime_load_errors() + + reports = [ + _doctor_plugin_report(n, d, n, enabled, disabled, load_errors) + for n, d in installed + ] + + if name is not None: + facts = reports[0] + console.print() + console.print(f"[bold]{facts['name']}[/bold]") + console.print( + f"[dim]Manifest:[/dim] " + + ("[green]ok[/green]" if facts["manifest_ok"] + else "[red]missing/invalid plugin.yaml[/red]") + ) + console.print(f"[dim]Status:[/dim] {facts['status']}") + if facts["load_error"]: + console.print(f"[dim]Load error:[/dim] [red]{facts['load_error']}[/red]") + if facts["missing_env"]: + console.print( + f"[dim]Missing env:[/dim] [yellow]{', '.join(facts['missing_env'])}[/yellow]" + ) + if facts["requires_hermes"]: + console.print(f"[dim]Requires:[/dim] hermes {facts['requires_hermes']}") + if facts["catalog"]: + console.print(f"[dim]Provenance:[/dim] {facts['catalog']}") + console.print(f"[dim]Pin:[/dim] {facts['pin']}") + if facts["removed"]: + console.print( + f"[red bold]✗ REMOVED from catalog: {facts['removed']}[/red bold]" + ) + console.print() + return + + table = Table(title="Plugin Doctor", show_lines=False) + table.add_column("Name", style="bold") + table.add_column("Manifest") + table.add_column("Status") + table.add_column("Issues") + table.add_column("Catalog", style="dim") + + for facts in reports: + issues: list[str] = [] + if facts["load_error"]: + issues.append(f"[red]{facts['load_error']}[/red]") + if facts["missing_env"]: + issues.append( + f"[yellow]missing env: {', '.join(facts['missing_env'])}[/yellow]" + ) + if facts["requires_hermes"] and "✗" in facts["requires_hermes"]: + issues.append(f"[red]requires hermes {facts['requires_hermes']}[/red]") + if facts["removed"]: + issues.append( + f"[red bold]REMOVED from catalog: {facts['removed']}[/red bold]" + ) + if facts["pin"] and "behind" in facts["pin"]: + issues.append(f"[yellow]{facts['pin']}[/yellow]") + table.add_row( + facts["name"], + "[green]ok[/green]" if facts["manifest_ok"] else "[red]bad[/red]", + facts["status"], + "\n".join(issues) or "[dim]—[/dim]", + facts["catalog"] or "[dim]—[/dim]", + ) + + console.print() + console.print(table) + console.print() + + # --------------------------------------------------------------------------- # Provider plugin discovery helpers # --------------------------------------------------------------------------- @@ -2083,7 +2665,18 @@ def plugins_command(args) -> None: args.identifier, force=getattr(args, "force", False), enable=enable_arg, + allow_removed=getattr(args, "allow_removed", False), ) + elif action == "search": + cmd_search(getattr(args, "query", "") or "") + elif action == "browse": + cmd_browse() + elif action == "info": + cmd_info(args.name) + elif action == "validate": + cmd_validate(args.path, as_json=getattr(args, "json", False)) + elif action == "doctor": + cmd_doctor(getattr(args, "name", None)) elif action == "update": cmd_update(args.name) elif action in {"remove", "rm", "uninstall"}: diff --git a/hermes_cli/subcommands/plugins.py b/hermes_cli/subcommands/plugins.py index 5355fbec3429..636b22ed72b1 100644 --- a/hermes_cli/subcommands/plugins.py +++ b/hermes_cli/subcommands/plugins.py @@ -42,6 +42,51 @@ def build_plugins_parser(subparsers, *, cmd_plugins: Callable) -> None: action="store_true", help="Install disabled (skip confirmation prompt); enable later with `hermes plugins enable `", ) + plugins_install.add_argument( + "--allow-removed", + action="store_true", + help="DANGEROUS: bypass the catalog removed-plugin blocklist check", + ) + + plugins_search = plugins_subparsers.add_parser( + "search", help="Search the Hermes plugin catalog" + ) + plugins_search.add_argument( + "query", + nargs="?", + default="", + help="Substring to match against entry names, descriptions, and tools", + ) + + plugins_subparsers.add_parser( + "browse", help="Browse every plugin catalog entry" + ) + + plugins_info = plugins_subparsers.add_parser( + "info", help="Show full catalog details for an entry" + ) + plugins_info.add_argument("name", help="Catalog entry name") + + plugins_validate = plugins_subparsers.add_parser( + "validate", + help="Validate a plugin directory for catalog admission (CI gate)", + ) + plugins_validate.add_argument("path", help="Path to the plugin directory") + plugins_validate.add_argument( + "--json", + action="store_true", + help="Print machine-readable JSON (for CI)", + ) + + plugins_doctor = plugins_subparsers.add_parser( + "doctor", help="Diagnose installed plugins" + ) + plugins_doctor.add_argument( + "name", + nargs="?", + default=None, + help="Plugin name to inspect in detail (default: all installed)", + ) plugins_update = plugins_subparsers.add_parser( "update", help="Pull latest changes for an installed plugin" From 373ad70f4ca4391f06cb26bb481f942b594232fc Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 22 Jul 2026 08:14:51 -0700 Subject: [PATCH 4/4] test(plugins): cover catalog CLI surface and plugin validation Behavior contracts for catalog-name install resolution (ref pin + sidecar), custom-URL banner, --allow-removed wiring, catalog-pin updates, list annotations, live-index fetch/fallback/TTL cache, search/browse/info rendering, doctor, argparse dispatch, and the validate checks incl. undeclared-capability diffs, crash containment, and built-in tool collisions. --- tests/hermes_cli/test_plugin_validate.py | 214 +++++++ tests/hermes_cli/test_plugins_cmd_catalog.py | 626 +++++++++++++++++++ 2 files changed, 840 insertions(+) create mode 100644 tests/hermes_cli/test_plugin_validate.py create mode 100644 tests/hermes_cli/test_plugins_cmd_catalog.py diff --git a/tests/hermes_cli/test_plugin_validate.py b/tests/hermes_cli/test_plugin_validate.py new file mode 100644 index 000000000000..d1051cdbc73c --- /dev/null +++ b/tests/hermes_cli/test_plugin_validate.py @@ -0,0 +1,214 @@ +"""Tests for ``hermes plugins validate`` (hermes_cli/plugin_validate.py). + +Static manifest checks + subprocess-isolated capability probing against a +recording stub context. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml + +import hermes_cli.plugins_cmd as plugins_cmd +from hermes_cli.plugin_validate import validate_plugin_dir + + +def _make_plugin( + tmp_path: Path, + *, + manifest: dict, + init_py: str = "def register(ctx):\n pass\n", +) -> Path: + d = tmp_path / manifest.get("name", "fixture-plugin") + d.mkdir(parents=True, exist_ok=True) + (d / "plugin.yaml").write_text(yaml.safe_dump(manifest), encoding="utf-8") + (d / "__init__.py").write_text(init_py, encoding="utf-8") + return d + + +BASE_MANIFEST = { + "name": "fixture-plugin", + "version": "1.0.0", + "description": "A fixture plugin.", +} + + +class TestStaticChecks: + def test_valid_plugin_passes(self, tmp_path): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + report = validate_plugin_dir(d) + assert report.ok + assert report.exit_code == 0 + + def test_missing_manifest_fails(self, tmp_path): + d = tmp_path / "empty-plugin" + d.mkdir() + report = validate_plugin_dir(d) + assert not report.ok + assert report.exit_code == 1 + assert any("plugin.yaml" in f for f in report.failures) + + def test_missing_required_fields_fail(self, tmp_path): + d = _make_plugin(tmp_path, manifest={"name": "fixture-plugin"}) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "version" in joined + assert "description" in joined + + def test_bad_requires_hermes_spec_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_hermes=">=not.a.version") + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("requires_hermes" in f for f in report.failures) + + def test_good_requires_hermes_spec_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_hermes=">=0.1, <99") + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_invalid_config_section_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, config=[{"prompt": "no key here"}]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("config" in f for f in report.failures) + + def test_valid_config_section_passes(self, tmp_path): + manifest = dict( + BASE_MANIFEST, + config=[ + {"key": "endpoint", "prompt": "Endpoint?", "type": "str"}, + {"key": "token", "secret": True, "type": "str"}, + ], + ) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_lower_snake_requires_env_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_env=["lower_case_bad"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("requires_env" in f for f in report.failures) + + def test_upper_snake_requires_env_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_env=["MY_API_KEY_2"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_rich_requires_env_dict_entries_accepted(self, tmp_path): + manifest = dict( + BASE_MANIFEST, + requires_env=[{"name": "MY_KEY", "description": "key"}], + ) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + +class TestCapabilityProbe: + def test_undeclared_tool_registration_fails_with_diff(self, tmp_path): + init = ( + "def register(ctx):\n" + " ctx.register_tool('sneaky_tool', 'sneaky', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "sneaky_tool" in joined + assert "undeclared" in joined.lower() + + def test_declared_and_registered_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["good_tool"]) + init = ( + "def register(ctx):\n" + " ctx.register_tool('good_tool', 'good', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=manifest, init_py=init) + report = validate_plugin_dir(d) + assert report.ok + + def test_declared_but_not_registered_warns(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["phantom_tool"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok # warn, not fail + assert any("phantom_tool" in w for w in report.warnings) + + def test_undeclared_hook_registration_fails(self, tmp_path): + init = ( + "def register(ctx):\n" + " ctx.register_hook('pre_tool_call', lambda **kw: None)\n" + ) + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + assert any("pre_tool_call" in f for f in report.failures) + + def test_crashing_register_is_contained(self, tmp_path): + init = "def register(ctx):\n raise RuntimeError('boom')\n" + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) # must not raise / kill the CLI + assert not report.ok + assert any("boom" in f or "register()" in f for f in report.failures) + + def test_import_time_os_exit_is_contained(self, tmp_path): + init = "import os\nos._exit(7)\n" + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + + def test_builtin_tool_collision_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["terminal"]) + init = ( + "def register(ctx):\n" + " ctx.register_tool('terminal', 'shadow', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=manifest, init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "terminal" in joined + assert "built-in" in joined + + +class TestCmdValidate: + def test_cmd_validate_exit_zero_on_pass(self, tmp_path, capsys): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d)) + assert e.value.code == 0 + out = capsys.readouterr().out + assert "✓" in out + + def test_cmd_validate_exit_one_on_fail(self, tmp_path, capsys): + d = tmp_path / "not-a-plugin" + d.mkdir() + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d)) + assert e.value.code == 1 + out = capsys.readouterr().out + assert "✗" in out + + def test_cmd_validate_json_output(self, tmp_path, capsys): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d), as_json=True) + assert e.value.code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert "checks" in payload + + def test_cmd_validate_missing_dir_fails(self, tmp_path, capsys): + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(tmp_path / "ghost")) + assert e.value.code == 1 diff --git a/tests/hermes_cli/test_plugins_cmd_catalog.py b/tests/hermes_cli/test_plugins_cmd_catalog.py new file mode 100644 index 000000000000..809b1de3a61e --- /dev/null +++ b/tests/hermes_cli/test_plugins_cmd_catalog.py @@ -0,0 +1,626 @@ +"""Tests for the catalog-driven ``hermes plugins`` CLI surface. + +Covers: catalog-name install resolution (pinned ref + provenance sidecar), +custom-URL banner, --allow-removed wiring, catalog-pin updates, list +annotations, live-index fetch/fallback/TTL, search/browse/info rendering, +doctor, and argparse dispatch. +""" + +from __future__ import annotations + +import argparse +import json +import types +from pathlib import Path + +import pytest +import yaml + +import hermes_cli.plugin_catalog as plugin_catalog +import hermes_cli.plugins_cmd as plugins_cmd +from hermes_constants import get_hermes_home + +SHA_A = "a" * 40 +SHA_B = "b" * 40 + + +# ── Helpers / fixtures ───────────────────────────────────────────────────── + + +def _write_entry(catalog_dir: Path, name: str, **overrides) -> Path: + data = { + "name": name, + "repo": f"https://github.com/example/{name}", + "sha": SHA_A, + "description": f"Test entry {name}.", + "maintainer": "Example", + } + data.update(overrides) + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / f"{name}.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return path + + +def _write_removed(catalog_dir: Path, removed: list) -> Path: + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / "removed.yaml" + path.write_text(yaml.safe_dump({"removed": removed}), encoding="utf-8") + return path + + +def _install_user_plugin(name: str, *, sidecar: dict | None = None) -> Path: + """Create a fake installed plugin under the per-test HERMES_HOME.""" + d = get_hermes_home() / "plugins" / name + d.mkdir(parents=True, exist_ok=True) + (d / "plugin.yaml").write_text( + yaml.safe_dump( + {"name": name, "version": "1.0.0", "description": f"{name} plugin"} + ), + encoding="utf-8", + ) + if sidecar is not None: + (d / ".hermes-catalog.json").write_text( + json.dumps(sidecar), encoding="utf-8" + ) + return d + + +@pytest.fixture() +def catalog_dir(tmp_path, monkeypatch): + d = tmp_path / "catalog" + d.mkdir() + monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(d)) + return d + + +@pytest.fixture() +def offline(monkeypatch): + """Force the live-index path to fall back to the in-tree catalog.""" + monkeypatch.setattr(plugin_catalog, "fetch_live_catalog", lambda **kw: None) + + +@pytest.fixture() +def fake_core(monkeypatch, tmp_path): + """Replace _install_plugin_core with a recording fake.""" + calls: list[dict] = [] + target = tmp_path / "fake-installed" + + def fake(identifier, *, force, ref=None, skip_removed_check=False): + target.mkdir(parents=True, exist_ok=True) + calls.append( + { + "identifier": identifier, + "force": force, + "ref": ref, + "skip_removed_check": skip_removed_check, + } + ) + return target, {"name": "my-entry"}, "my-entry" + + monkeypatch.setattr(plugins_cmd, "_install_plugin_core", fake) + return types.SimpleNamespace(calls=calls, target=target) + + +# ── Catalog-name install ─────────────────────────────────────────────────── + + +class TestCatalogInstall: + def test_catalog_name_resolves_to_pinned_repo( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_A) + plugins_cmd.cmd_install("my-entry", enable=False) + assert len(fake_core.calls) == 1 + call = fake_core.calls[0] + assert call["identifier"] == "https://github.com/example/my-entry" + assert call["ref"] == SHA_A + assert call["skip_removed_check"] is False + + def test_subdir_entry_uses_fragment_identifier( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", subdir="plugins/inner") + plugins_cmd.cmd_install("my-entry", enable=False) + assert fake_core.calls[0]["identifier"] == ( + "https://github.com/example/my-entry#plugins/inner" + ) + + def test_sidecar_written_with_provenance( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", tier="official") + plugins_cmd.cmd_install("my-entry", enable=False) + sidecar_path = fake_core.target / ".hermes-catalog.json" + assert sidecar_path.is_file() + sidecar = json.loads(sidecar_path.read_text(encoding="utf-8")) + assert sidecar["catalog_name"] == "my-entry" + assert sidecar["repo"] == "https://github.com/example/my-entry" + assert sidecar["sha"] == SHA_A + assert sidecar["tier"] == "official" + assert sidecar["installed_at"] + + def test_capability_summary_and_tier_shown( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry( + catalog_dir, + "my-entry", + tier="official", + capabilities={"provides_tools": ["cool_tool"]}, + ) + plugins_cmd.cmd_install("my-entry", enable=False) + out = capsys.readouterr().out + assert "official" in out + assert "cool_tool" in out + + def test_unknown_catalog_like_name_errors( + self, catalog_dir, offline, fake_core, capsys + ): + with pytest.raises(SystemExit): + plugins_cmd.cmd_install("nonexistent-entry", enable=False) + out = capsys.readouterr().out + assert "search" in out + assert not fake_core.calls + + def test_custom_url_gets_unreviewed_banner( + self, catalog_dir, offline, fake_core, capsys + ): + plugins_cmd.cmd_install( + "https://github.com/foo/bar.git", enable=False + ) + out = capsys.readouterr().out + assert "custom (unreviewed) source" in out + # Custom installs never get a ref pin. + assert fake_core.calls[0]["ref"] is None + + def test_allow_removed_passes_skip_flag_and_warns( + self, catalog_dir, offline, fake_core, capsys + ): + plugins_cmd.cmd_install( + "https://github.com/foo/bar.git", + enable=False, + allow_removed=True, + ) + out = capsys.readouterr().out + assert fake_core.calls[0]["skip_removed_check"] is True + assert "removed" in out.lower() + + +# ── Catalog update ───────────────────────────────────────────────────────── + + +class TestCatalogUpdate: + def test_update_reinstalls_at_new_pin( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_B) + target = _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd.cmd_update("my-entry") + assert len(fake_core.calls) == 1 + call = fake_core.calls[0] + assert call["ref"] == SHA_B + assert call["force"] is True + out = capsys.readouterr().out + assert SHA_A[:8] in out + assert SHA_B[:8] in out + # Sidecar refreshed to the new pin (written into the reinstall target). + sidecar = json.loads( + (fake_core.target / ".hermes-catalog.json").read_text( + encoding="utf-8" + ) + ) + assert sidecar["sha"] == SHA_B + assert target.exists() or True # target replaced by reinstall + + def test_update_already_at_pin_is_noop( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_A) + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd.cmd_update("my-entry") + out = capsys.readouterr().out + assert "already at catalog pin" in out + assert not fake_core.calls + + def test_update_preserves_enabled_state( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_B) + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd._save_enabled_set({"my-entry"}) + plugins_cmd.cmd_update("my-entry") + assert "my-entry" in plugins_cmd._get_enabled_set() + + def test_update_without_sidecar_keeps_git_flow( + self, catalog_dir, offline, fake_core, capsys + ): + _install_user_plugin("plain-git-plugin") # no sidecar, no .git + with pytest.raises(SystemExit): + plugins_cmd.cmd_update("plain-git-plugin") + out = capsys.readouterr().out + assert "not installed from git" in out + assert not fake_core.calls + + def test_update_entry_gone_from_catalog_errors( + self, catalog_dir, offline, fake_core, capsys + ): + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + with pytest.raises(SystemExit): + plugins_cmd.cmd_update("my-entry") + out = capsys.readouterr().out + assert "no longer in the catalog" in out + + +# ── List annotations ─────────────────────────────────────────────────────── + + +class TestListAnnotations: + def test_json_includes_catalog_annotation( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "cat-plugin", + sidecar={ + "catalog_name": "cat-plugin", + "repo": "https://github.com/example/cat-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + args = argparse.Namespace(json=True) + plugins_cmd.cmd_list(args) + payload = json.loads(capsys.readouterr().out) + row = next(p for p in payload if p["name"] == "cat-plugin") + assert row["catalog"] == f"catalog:official@{SHA_A[:8]}" + + def test_removed_plugin_flagged(self, catalog_dir, offline, capsys): + _install_user_plugin("evil-plugin") + _write_removed( + catalog_dir, + [{"name": "evil-plugin", "reason": "exfiltrated env vars"}], + ) + plugins_cmd.cmd_list(argparse.Namespace()) + out = capsys.readouterr().out + assert "REMOVED from catalog" in out + assert "exfiltrated env vars" in out + + +# ── Live index fetch ─────────────────────────────────────────────────────── + + +class _FakeResp: + def __init__(self, *, json_data=None, text=""): + self._json = json_data + self.text = text + + def raise_for_status(self): + pass + + def json(self): + return self._json + + +def _fake_httpx_get(listing, files, counter): + def fake_get(url, **kwargs): + counter.append(url) + if "api.github.com" in url: + return _FakeResp(json_data=listing) + fname = url.rsplit("/", 1)[-1] + return _FakeResp(text=files[fname]) + + return fake_get + + +class TestLiveIndex: + def _remote_entry_yaml(self, name, sha=SHA_B): + return yaml.safe_dump( + { + "name": name, + "repo": f"https://github.com/example/{name}", + "sha": sha, + "description": f"Remote entry {name}.", + "maintainer": "Example", + } + ) + + def test_live_fetch_populates_cache_and_entries( + self, catalog_dir, monkeypatch + ): + _write_entry(catalog_dir, "local-entry") + listing = [ + { + "name": "remote-entry.yaml", + "download_url": "https://raw.example/remote-entry.yaml", + }, + ] + files = {"remote-entry.yaml": self._remote_entry_yaml("remote-entry")} + counter: list[str] = [] + monkeypatch.setattr( + "httpx.get", _fake_httpx_get(listing, files, counter) + ) + entries = plugin_catalog.load_catalog_live() + names = [e.name for e in entries] + assert names == ["remote-entry"] + cache = get_hermes_home() / "cache" / "plugin-catalog" + assert (cache / "remote-entry.yaml").is_file() + + def test_network_failure_falls_back_to_in_tree( + self, catalog_dir, monkeypatch + ): + _write_entry(catalog_dir, "local-entry") + + def boom(url, **kwargs): + raise OSError("no network") + + monkeypatch.setattr("httpx.get", boom) + entries = plugin_catalog.load_catalog_live() + assert [e.name for e in entries] == ["local-entry"] + + def test_ttl_cache_skips_refetch(self, catalog_dir, monkeypatch): + listing = [ + { + "name": "remote-entry.yaml", + "download_url": "https://raw.example/remote-entry.yaml", + }, + ] + files = {"remote-entry.yaml": self._remote_entry_yaml("remote-entry")} + counter: list[str] = [] + monkeypatch.setattr( + "httpx.get", _fake_httpx_get(listing, files, counter) + ) + plugin_catalog.load_catalog_live() + first_count = len(counter) + assert first_count >= 2 # listing + file + + # Second call within TTL must not hit the network at all — even if + # the network is now broken. + def boom(url, **kwargs): + raise AssertionError("network hit despite fresh cache") + + monkeypatch.setattr("httpx.get", boom) + entries = plugin_catalog.load_catalog_live() + assert [e.name for e in entries] == ["remote-entry"] + + +# ── search / browse / info ───────────────────────────────────────────────── + + +class TestSearchBrowseInfo: + def test_search_filters_entries(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_entry(catalog_dir, "beta-entry") + plugins_cmd.cmd_search("alpha") + out = capsys.readouterr().out + assert "alpha-entry" in out + assert "beta-entry" not in out + + def test_browse_lists_all(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_entry(catalog_dir, "beta-entry") + plugins_cmd.cmd_browse() + out = capsys.readouterr().out + assert "alpha-entry" in out + assert "beta-entry" in out + + def test_search_no_results_message(self, catalog_dir, offline, capsys): + plugins_cmd.cmd_search("zzz-nothing") + out = capsys.readouterr().out + assert "No catalog entries" in out + + def test_info_shows_full_detail(self, catalog_dir, offline, capsys): + _write_entry( + catalog_dir, + "alpha-entry", + tier="official", + requires_hermes=">=0.19", + docs_url="https://example.com/docs", + platforms=["linux"], + capabilities={ + "provides_tools": ["cool_tool"], + "requires_env": ["ALPHA_KEY"], + }, + ) + plugins_cmd.cmd_info("alpha-entry") + out = capsys.readouterr().out + assert SHA_A in out + assert "official" in out + assert "cool_tool" in out + assert "ALPHA_KEY" in out + assert ">=0.19" in out + assert "hermes plugins install alpha-entry" in out + + def test_info_unknown_entry_exits(self, catalog_dir, offline, capsys): + with pytest.raises(SystemExit): + plugins_cmd.cmd_info("ghost-entry") + + def test_info_warns_when_removed(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_removed( + catalog_dir, + [{"name": "alpha-entry", "reason": "bad actor"}], + ) + plugins_cmd.cmd_info("alpha-entry") + out = capsys.readouterr().out + assert "REMOVED" in out + assert "bad actor" in out + + +# ── doctor ───────────────────────────────────────────────────────────────── + + +class TestDoctor: + @pytest.fixture(autouse=True) + def _no_runtime_scan(self, monkeypatch): + monkeypatch.setattr( + plugins_cmd, "_runtime_load_errors", lambda: {} + ) + + def test_doctor_table_lists_installed_plugin( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "doc-plugin", + sidecar={ + "catalog_name": "doc-plugin", + "repo": "https://github.com/example/doc-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + _write_entry(catalog_dir, "doc-plugin", sha=SHA_A, tier="official") + plugins_cmd.cmd_doctor() + out = capsys.readouterr().out + assert "doc-plugin" in out + assert "official" in out + + def test_doctor_detail_flags_pin_mismatch( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "doc-plugin", + sidecar={ + "catalog_name": "doc-plugin", + "repo": "https://github.com/example/doc-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + _write_entry(catalog_dir, "doc-plugin", sha=SHA_B) + plugins_cmd.cmd_doctor("doc-plugin") + out = capsys.readouterr().out + assert "doc-plugin" in out + assert "behind catalog pin" in out or "pin mismatch" in out + + def test_doctor_flags_removed_plugin(self, catalog_dir, offline, capsys): + _install_user_plugin("evil-plugin") + _write_removed( + catalog_dir, + [{"name": "evil-plugin", "reason": "exfiltrated env vars"}], + ) + plugins_cmd.cmd_doctor("evil-plugin") + out = capsys.readouterr().out + assert "REMOVED" in out + + def test_doctor_unknown_plugin_exits(self, catalog_dir, offline, capsys): + with pytest.raises(SystemExit): + plugins_cmd.cmd_doctor("no-such-plugin") + + +# ── argparse dispatch ────────────────────────────────────────────────────── + + +class TestDispatch: + def _dispatch(self, monkeypatch, action, **attrs): + recorded = {} + + def record(fn_name): + def _rec(*args, **kwargs): + recorded["fn"] = fn_name + recorded["args"] = args + recorded["kwargs"] = kwargs + + return _rec + + for fn in ( + "cmd_search", + "cmd_browse", + "cmd_info", + "cmd_validate", + "cmd_doctor", + "cmd_install", + ): + monkeypatch.setattr(plugins_cmd, fn, record(fn)) + ns = argparse.Namespace(plugins_action=action, **attrs) + plugins_cmd.plugins_command(ns) + return recorded + + def test_search_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "search", query="foo") + assert rec["fn"] == "cmd_search" + assert "foo" in rec["args"] or rec["kwargs"].get("query") == "foo" + + def test_browse_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "browse") + assert rec["fn"] == "cmd_browse" + + def test_info_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "info", name="foo") + assert rec["fn"] == "cmd_info" + + def test_validate_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "validate", path="/tmp/x", json=True) + assert rec["fn"] == "cmd_validate" + + def test_doctor_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "doctor", name=None) + assert rec["fn"] == "cmd_doctor" + + def test_install_allow_removed_dispatch(self, monkeypatch): + rec = self._dispatch( + monkeypatch, + "install", + identifier="x", + force=False, + enable=False, + no_enable=True, + allow_removed=True, + ) + assert rec["fn"] == "cmd_install" + assert rec["kwargs"].get("allow_removed") is True + + def test_parser_wires_new_subcommands(self): + from hermes_cli.subcommands.plugins import build_plugins_parser + + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="command") + build_plugins_parser(sub, cmd_plugins=lambda args: None) + for argv in ( + ["plugins", "search", "foo"], + ["plugins", "browse"], + ["plugins", "info", "foo"], + ["plugins", "validate", "/tmp/x", "--json"], + ["plugins", "doctor"], + ["plugins", "install", "foo", "--allow-removed"], + ): + args = parser.parse_args(argv) + assert args.plugins_action == argv[1]