mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-22 16:25:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Merge remote-tracking branch 'origin/feat/plugcat-cli' into feat/plugin-catalog

Browse source
This commit is contained in:
Teknium 2026-07-22 08:15:38 -07:00
commit 741d06445d
No known key found for this signature in database
6 changed files with 2030 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

113
hermes_cli/plugin_catalog.py
View file

@ -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 ``<hermes_home>/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 ─────────────────────────────────────────────────────────

434
hermes_cli/plugin_validate.py Normal file
View file

@ -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

607
hermes_cli/plugins_cmd.py
View file

@ -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:<tier>@<shaShort>`` 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 <name>")
console.print("[dim]Install:[/dim] hermes plugins install <name>")
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"}:

45
hermes_cli/subcommands/plugins.py
View file

@ -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 <name>`",
)
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"

214
tests/hermes_cli/test_plugin_validate.py Normal file
View file

@ -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

626
tests/hermes_cli/test_plugins_cmd_catalog.py Normal file
View file

@ -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]