Compare commits

..
Author SHA1 Message Date
Teknium 8dd07bd517 feat(ci): plugin-validate reusable action + plugin-catalog admission gate
- scripts/validate_plugin_catalog.py: standalone stdlib+pyyaml structural
  validator for plugin-catalog entries and removed.yaml (no hermes install
  needed; runtime twin of hermes_cli/plugin_catalog.py). --json support,
  unknown top-level keys warn instead of failing for forward compat.
- .github/actions/plugin-validate: composite action plugin authors drop
  into their own repo's CI — installs hermes-agent from a chosen ref and
  runs 'hermes plugins validate <path>'.
- .github/workflows/plugin-catalog-ci.yml: admission gate on PRs touching
  plugin-catalog/** — structural job plus pinned-source job that clones
  each changed entry's repo, hard-fails on unreachable pinned sha
  (supply-chain gate), and validates the plugin at that exact commit.
2026-07-22 07:22:43 -07:00
15 changed files with 749 additions and 3246 deletions
@@ -0,0 +1,56 @@
name: Hermes Plugin Validate
description: >-
Validate a Hermes Agent plugin (plugin.yaml manifest schema AND
declared-vs-actually-registered capabilities) using
`hermes plugins validate`. Drop this into your plugin repo's CI:
- uses: actions/checkout@<sha>
- uses: NousResearch/hermes-agent/.github/actions/plugin-validate@main
with:
path: .
The caller's job owns checkout; this action installs Python + hermes-agent
(git install — a supported CI-context install route) and runs the
validator against your plugin directory.
inputs:
path:
description: Path to the plugin directory (containing plugin.yaml).
default: "."
hermes-ref:
description: hermes-agent git ref (branch/tag/sha) to install and validate with.
default: "main"
runs:
using: composite
steps:
- name: Set up Python
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: "3.11"
- name: Install hermes-agent
shell: bash
env:
_HERMES_REF: ${{ inputs.hermes-ref }}
run: |
set -euo pipefail
# CI-context install from git; the ref lets plugin authors validate
# against a pinned hermes release instead of main.
pip install "git+https://github.com/NousResearch/hermes-agent@${_HERMES_REF}"
- name: Validate plugin
shell: bash
env:
_PLUGIN_PATH: ${{ inputs.path }}
run: |
set -uo pipefail
# `hermes plugins validate` checks the plugin.yaml manifest schema
# and loads the plugin in a scratch subprocess to verify that the
# capabilities it DECLARES match what it actually registers.
if hermes plugins validate "$_PLUGIN_PATH"; then
echo "✅ PASS: plugin at '$_PLUGIN_PATH' validated cleanly"
else
echo "❌ FAIL: plugin at '$_PLUGIN_PATH' failed validation (see output above)"
exit 1
fi
+140
View File
@@ -0,0 +1,140 @@
name: Plugin Catalog CI
# Admission gate for plugin-catalog entries. Fires ONLY on PRs touching
# plugin-catalog/** so it can never go red on unrelated PRs.
#
# Two gates:
# structural — cheap schema check, no hermes install needed
# pinned-source-validate — supply-chain gate: the pinned sha MUST be
# reachable in the entry's repo, and the plugin
# at that exact commit must pass
# `hermes plugins validate`.
on:
pull_request:
paths:
- "plugin-catalog/**"
permissions:
contents: read
jobs:
structural:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: "3.11"
- name: Install PyYAML
uses: ./.github/actions/retry
with:
command: pip install pyyaml==6.0.2
- name: Validate catalog files (structural)
run: |
set -euo pipefail
# Validating the whole directory is simpler than diffing and keeps
# the invariant that EVERYTHING in plugin-catalog/ stays valid.
python3 scripts/validate_plugin_catalog.py plugin-catalog/
pinned-source-validate:
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
with:
fetch-depth: 0 # need the merge-base to diff changed catalog files
- uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
with:
python-version: "3.11"
- name: Find changed catalog entries
id: changed
run: |
set -euo pipefail
MERGE_BASE=$(git merge-base "origin/${{ github.base_ref }}" HEAD)
# Added + modified entry files only; deletions and removed.yaml
# have nothing to clone.
CHANGED=$(git diff --name-only --diff-filter=AM "$MERGE_BASE"...HEAD \
-- 'plugin-catalog/*.yaml' 'plugin-catalog/*.yml' \
| grep -v '/removed\.yaml$' || true)
echo "Changed catalog entries:"
echo "${CHANGED:-<none>}"
{
echo 'files<<__EOF__'
echo "$CHANGED"
echo '__EOF__'
} >> "$GITHUB_OUTPUT"
- name: Install hermes-agent from the PR's own checkout
if: steps.changed.outputs.files != ''
uses: ./.github/actions/retry
with:
command: pip install -e .
- name: Clone each entry at its pinned sha and validate
if: steps.changed.outputs.files != ''
env:
CHANGED_FILES: ${{ steps.changed.outputs.files }}
run: |
set -euo pipefail
FAILED=0
while IFS= read -r entry; do
[ -z "$entry" ] && continue
echo "::group::validate $entry"
# Parse repo / sha / subdir from the entry yaml.
eval "$(python3 - "$entry" <<'PYEOF'
import shlex
import sys
import yaml
with open(sys.argv[1], encoding="utf-8") as fh:
data = yaml.safe_load(fh) or {}
print(f"REPO={shlex.quote(str(data.get('repo', '')))}")
print(f"SHA={shlex.quote(str(data.get('sha', '')))}")
print(f"SUBDIR={shlex.quote(str(data.get('subdir', '') or ''))}")
PYEOF
)"
echo "repo=$REPO sha=$SHA subdir=$SUBDIR"
CLONE_DIR=$(mktemp -d)
# Full clone (no --depth 1): the pinned sha may not be the branch tip.
if ! git clone "$REPO" "$CLONE_DIR"; then
echo "::error file=$entry::clone failed for $REPO"
FAILED=1; echo "::endgroup::"; continue
fi
# SUPPLY-CHAIN GATE: the pinned sha must be reachable in the repo.
if ! git -C "$CLONE_DIR" checkout --detach "$SHA"; then
echo "::error file=$entry::pinned sha $SHA is not reachable in $REPO"
FAILED=1; echo "::endgroup::"; continue
fi
PLUGIN_DIR="$CLONE_DIR${SUBDIR:+/$SUBDIR}"
if [ ! -f "$PLUGIN_DIR/plugin.yaml" ]; then
echo "::error file=$entry::no plugin.yaml at subdir '$SUBDIR' of $REPO@$SHA"
FAILED=1; echo "::endgroup::"; continue
fi
# Manifest schema + declared-vs-registered capability check.
if hermes plugins validate "$PLUGIN_DIR"; then
echo "✅ PASS: $entry"
else
echo "::error file=$entry::hermes plugins validate failed"
FAILED=1
fi
echo "::endgroup::"
done <<< "$CHANGED_FILES"
if [ "$FAILED" -ne 0 ]; then
echo "❌ FAIL: one or more catalog entries failed pinned-source validation"
exit 1
fi
echo "✅ PASS: all changed catalog entries validated at their pinned shas"
-416
View File
@@ -1,416 +0,0 @@
"""Plugin catalog — curated, Nous-approved Hermes plugins shipped with the repo.
Mirrors the ``optional-mcps/`` MCP-catalog pattern (see
:mod:`hermes_cli.mcp_catalog`): each catalog entry is a single YAML file under
the in-tree ``plugin-catalog/`` directory, pinned to an exact 40-character
commit SHA. Users discover entries via ``hermes plugins catalog`` /
``hermes plugins search`` and install them with
``hermes plugins install <name>``, which clones the pinned commit.
Catalog policy (see plugin-catalog/README.md for the full admission policy):
- Entries are added only by merging a PR into hermes-agent — presence in the
``plugin-catalog/`` directory is the human-merged approval gate.
- Every entry pins an exact 40-hex commit SHA. SHA bumps are new PRs,
re-reviewed as diffs. The pinned release should be at least 2 weeks old at
pin time, mirroring the optional-mcps supply-chain rules.
- ``plugin-catalog/removed.yaml`` is the blocklist: entries pulled from the
catalog for security or policy reasons are recorded there so installs of
the same name/repo are refused with the recorded reason.
"""
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
import yaml
logger = logging.getLogger(__name__)
CATALOG_TIERS = ("official", "community")
_SHA_RE = re.compile(r"^[0-9a-f]{40}$")
_NAME_RE = re.compile(r"^[a-z0-9_-]{1,64}$")
# ─── Data classes ────────────────────────────────────────────────────────────
@dataclass
class RemovedEntry:
name: str
repo: str = ""
reason: str = ""
date: str = "" # ISO date string
@dataclass
class CatalogCapabilities:
provides_tools: List[str] = field(default_factory=list)
provides_hooks: List[str] = field(default_factory=list)
provides_middleware: List[str] = field(default_factory=list)
requires_env: List[str] = field(default_factory=list)
@dataclass
class PluginCatalogEntry:
name: str # catalog key, [a-z0-9_-]{1,64}
repo: str # https:// git URL
sha: str # 40-hex pinned commit — MANDATORY, validated
description: str
maintainer: str
tier: str = "community" # one of CATALOG_TIERS
requires_hermes: str = "" # e.g. ">=0.19" (optional)
subdir: str = "" # optional path within the repo
docs_url: str = ""
platforms: List[str] = field(default_factory=list) # empty = all OSes
capabilities: CatalogCapabilities = field(default_factory=CatalogCapabilities)
# ─── Directory resolution ────────────────────────────────────────────────────
def get_catalog_dir() -> Path:
"""Return the ``plugin-catalog/`` directory shipped with this checkout.
``HERMES_PLUGIN_CATALOG_DIR`` overrides the location for tests only —
read via ``os.getenv`` at call time so monkeypatched values take effect.
"""
override = os.getenv("HERMES_PLUGIN_CATALOG_DIR", "").strip()
if override:
return Path(override)
return Path(__file__).resolve().parent.parent / "plugin-catalog"
# ─── Loading / validation ────────────────────────────────────────────────────
def _str_list(raw: Any) -> List[str]:
"""Coerce a YAML value into a list of strings (drop non-strings)."""
if not isinstance(raw, list):
return []
return [str(item) for item in raw if isinstance(item, (str, int, float))]
def _parse_entry(path: Path) -> Optional[PluginCatalogEntry]:
"""Parse and validate one catalog YAML file.
Returns ``None`` (after logging a warning) on any validation failure —
the loader never raises for a bad entry.
"""
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except Exception as exc:
logger.warning("Plugin catalog: failed to read %s: %s", path, exc)
return None
if not isinstance(data, dict):
logger.warning("Plugin catalog: %s: entry must be a mapping", path)
return None
name = str(data.get("name") or "")
if not _NAME_RE.match(name):
logger.warning(
"Plugin catalog: %s: invalid name %r (must match [a-z0-9_-]{1,64})",
path, name,
)
return None
repo = str(data.get("repo") or "")
if not repo.startswith("https://"):
logger.warning(
"Plugin catalog: %s: repo must be an https:// URL (got %r)",
path, repo,
)
return None
sha = str(data.get("sha") or "").strip().lower()
if not _SHA_RE.match(sha):
logger.warning(
"Plugin catalog: %s: sha must be a full 40-character hex commit "
"SHA (got %r)", path, data.get("sha"),
)
return None
tier = str(data.get("tier") or "community")
if tier not in CATALOG_TIERS:
logger.warning(
"Plugin catalog: %s: tier must be one of %s (got %r)",
path, "/".join(CATALOG_TIERS), tier,
)
return None
caps_raw = data.get("capabilities") or {}
if not isinstance(caps_raw, dict):
caps_raw = {}
capabilities = CatalogCapabilities(
provides_tools=_str_list(caps_raw.get("provides_tools")),
provides_hooks=_str_list(caps_raw.get("provides_hooks")),
provides_middleware=_str_list(caps_raw.get("provides_middleware")),
requires_env=_str_list(caps_raw.get("requires_env")),
)
return PluginCatalogEntry(
name=name,
repo=repo,
sha=sha,
description=str(data.get("description") or "").strip(),
maintainer=str(data.get("maintainer") or "").strip(),
tier=tier,
requires_hermes=str(data.get("requires_hermes") or "").strip(),
subdir=str(data.get("subdir") or "").strip(),
docs_url=str(data.get("docs_url") or "").strip(),
platforms=_str_list(data.get("platforms")),
capabilities=capabilities,
)
def load_catalog() -> List[PluginCatalogEntry]:
"""Return all valid catalog entries, sorted by name.
Parses every ``*.yaml`` in the catalog dir except ``removed.yaml``.
Invalid entries are skipped with a logged warning; this function never
raises for a malformed entry.
"""
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] = []
for path in sorted(root.glob("*.yaml")):
if path.name == "removed.yaml":
continue
entry = _parse_entry(path)
if entry is not None:
entries.append(entry)
return entries
def get_catalog_entry(name: str) -> Optional[PluginCatalogEntry]:
"""Look up a single catalog entry by name."""
for entry in load_catalog():
if entry.name == name:
return entry
return None
def search_catalog(query: str) -> List[PluginCatalogEntry]:
"""Case-insensitive substring search over name, description, and
declared tools. An empty query returns the whole 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
results: List[PluginCatalogEntry] = []
for entry in entries:
haystacks = [entry.name, entry.description]
haystacks.extend(entry.capabilities.provides_tools)
if any(q in h.lower() for h in haystacks):
results.append(entry)
return results
# ─── Removed / blocklist ─────────────────────────────────────────────────────
def _normalize_repo(url: str) -> str:
"""Normalize a repo URL for comparison (.git suffix and trailing slash
stripped, lowercased)."""
return url.strip().rstrip("/").removesuffix(".git").lower()
def load_removed_list() -> List[RemovedEntry]:
"""Load ``plugin-catalog/removed.yaml`` (the ``removed:`` list).
Missing or malformed files yield an empty list — never raises.
"""
path = get_catalog_dir() / "removed.yaml"
if not path.is_file():
return []
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
except Exception as exc:
logger.warning("Plugin catalog: failed to read %s: %s", path, exc)
return []
raw_list = data.get("removed") if isinstance(data, dict) else None
if not isinstance(raw_list, list):
return []
removed: List[RemovedEntry] = []
for raw in raw_list:
if not isinstance(raw, dict):
continue
name = str(raw.get("name") or "")
if not name:
continue
removed.append(
RemovedEntry(
name=name,
repo=str(raw.get("repo") or ""),
reason=str(raw.get("reason") or ""),
date=str(raw.get("date") or ""),
)
)
return removed
def find_removed(name_or_repo: str) -> Optional[RemovedEntry]:
"""Match *name_or_repo* against the removed blocklist.
Matches by exact catalog name OR by repo URL (normalized — ``.git``
suffix and trailing slashes are ignored).
"""
if not name_or_repo:
return None
candidate = name_or_repo.strip()
candidate_repo = _normalize_repo(candidate)
for entry in load_removed_list():
if candidate == entry.name:
return entry
if entry.repo and candidate_repo == _normalize_repo(entry.repo):
return entry
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 ─────────────────────────────────────────────────────────
def entry_capability_summary(entry: PluginCatalogEntry) -> str:
"""One-paragraph human summary of what an entry declares, shown at
install prompts so the user knows what they're granting."""
caps = entry.capabilities
parts: List[str] = []
if caps.provides_tools:
parts.append(f"registers tool(s): {', '.join(caps.provides_tools)}")
if caps.provides_hooks:
parts.append(f"hook(s): {', '.join(caps.provides_hooks)}")
if caps.provides_middleware:
parts.append(f"middleware: {', '.join(caps.provides_middleware)}")
if caps.requires_env:
parts.append(f"requires env var(s): {', '.join(caps.requires_env)}")
if not parts:
capability_text = "declares no tools, hooks, middleware, or env vars"
else:
capability_text = "; ".join(parts)
bits = [
f"{entry.name} ({entry.tier}, maintained by {entry.maintainer})",
]
if entry.description:
bits.append(entry.description)
bits.append(f"This plugin {capability_text}.")
if entry.platforms:
bits.append(f"Platforms: {', '.join(entry.platforms)}.")
if entry.requires_hermes:
bits.append(f"Requires Hermes {entry.requires_hermes}.")
return " ".join(bits)
-434
View File
@@ -1,434 +0,0 @@
"""``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
-169
View File
@@ -39,7 +39,6 @@ import importlib.util
import inspect
import logging
import os
import re
import sys
import threading
import types
@@ -80,93 +79,6 @@ class PluginToolOverrideError(PermissionError):
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Hermes version gate (manifest ``requires_hermes``)
# ---------------------------------------------------------------------------
_VERSION_COMPARATOR_RE = re.compile(r"^\s*(>=|<=|==|!=|>|<)\s*(.+?)\s*$")
def _running_hermes_version() -> str:
"""Return the running Hermes version string.
Prefers installed package metadata (matches ``hermes_cli/main.py``'s
version reporting), falling back to ``hermes_cli.__version__`` for
source checkouts, then ``"0.0.0"`` as a last resort.
"""
try:
return importlib.metadata.version("hermes-agent")
except Exception:
pass
try:
from hermes_cli import __version__
return __version__
except Exception:
return "0.0.0"
def _version_tuple(v: str) -> Optional[tuple]:
"""Parse ``major.minor.patch`` into a comparable tuple.
Leading ``v`` and pre-release/build metadata (``-rc1``, ``+abc``) are
stripped; missing segments default to 0. Returns ``None`` when any
segment is non-numeric.
"""
s = str(v).strip().lstrip("v")
s = re.split(r"[-+]", s, 1)[0]
parts = s.split(".")
while len(parts) < 3:
parts.append("0")
try:
return tuple(int(p) for p in parts[:3])
except ValueError:
return None
def _version_satisfies(spec: str, current: str) -> bool:
"""Return True when *current* satisfies *spec*.
*spec* supports ``>=``, ``>``, ``<=``, ``<``, ``==``, ``!=`` and
comma-separated combinations (all must hold). A bare version is treated
as ``>=``. Non-numeric version segments fall back to permissive True
(with a debug log) — no new dependency, so no full PEP 440 handling.
"""
if not spec or not spec.strip():
return True
cur = _version_tuple(current)
if cur is None:
logger.debug(
"requires_hermes: unparseable running version %r — allowing", current,
)
return True
for clause in spec.split(","):
clause = clause.strip()
if not clause:
continue
m = _VERSION_COMPARATOR_RE.match(clause)
if m:
op, target = m.group(1), m.group(2)
else:
op, target = ">=", clause
tgt = _version_tuple(target)
if tgt is None:
logger.debug(
"requires_hermes: unparseable version spec %r — allowing", clause,
)
continue
ok = {
">=": cur >= tgt,
"<=": cur <= tgt,
"==": cur == tgt,
"!=": cur != tgt,
">": cur > tgt,
"<": cur < tgt,
}[op]
if not ok:
return False
return True
# ---------------------------------------------------------------------------
# Plugin developer debug logging
# ---------------------------------------------------------------------------
@@ -400,16 +312,6 @@ class PluginManifest:
# category plugin at ``plugins/image_gen/openai/`` the key is
# ``image_gen/openai``. When empty, falls back to ``name``.
key: str = ""
# Minimum/exact Hermes version requirement, e.g. ``">=0.19"``. Empty =
# no requirement. Checked at load time; unsatisfied plugins are recorded
# with an error and skipped (no register() call, no traceback).
requires_hermes: str = ""
# Declared config keys from the manifest's ``config:`` section — a list
# of ``{key, prompt, type (str|bool|int), default, secret (bool)}``
# dicts. secret=true values are prompted into ~/.hermes/.env; secret
# =false values live under ``plugins.entries.<plugin_id>.<key>`` in
# config.yaml. Exposed to plugins via ``ctx.plugin_config``.
config_spec: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
@@ -484,40 +386,6 @@ class PluginContext:
except Exception:
return "default"
# -- declared plugin config ---------------------------------------------
@property
def plugin_config(self) -> Dict[str, Any]:
"""Return this plugin's effective config values.
Built from the manifest's ``config:`` spec defaults, overlaid with
whatever the operator set under ``plugins.entries.<plugin_id>`` in
config.yaml (config.yaml wins on key collision). Secret keys
(``secret: true``) are stored in ``~/.hermes/.env`` instead and are
NOT surfaced here — read them via ``os.environ``.
"""
merged: Dict[str, Any] = {}
for spec in self.manifest.config_spec or []:
key = spec.get("key")
if not key:
continue
if spec.get("secret"):
continue # secrets live in .env, never in config.yaml
if "default" in spec:
merged[key] = spec.get("default")
try:
from hermes_cli.config import load_config
cfg = load_config() or {}
except Exception:
cfg = {}
plugin_id = self.manifest.key or self.manifest.name
entries = (cfg.get("plugins") or {}).get("entries") or {}
entry = entries.get(plugin_id) or {}
if isinstance(entry, dict):
for key, value in entry.items():
merged[key] = value
return merged
# -- tool registration --------------------------------------------------
def register_tool(
@@ -1764,23 +1632,6 @@ class PluginManager:
"Parsed manifest: key=%s name=%s kind=%s source=%s path=%s",
key, name, kind, source, plugin_dir,
)
raw_config = data.get("config", [])
config_spec: List[Dict[str, Any]] = []
if isinstance(raw_config, list):
for item in raw_config:
if isinstance(item, dict) and item.get("key"):
config_spec.append(dict(item))
else:
logger.warning(
"Plugin %s: ignoring invalid config entry %r "
"(must be a mapping with a 'key')", key, item,
)
elif raw_config:
logger.warning(
"Plugin %s: 'config' must be a list of mappings; ignoring",
key,
)
return PluginManifest(
name=name,
version=str(data.get("version", "")),
@@ -1793,8 +1644,6 @@ class PluginManager:
path=str(plugin_dir),
kind=kind,
key=key,
requires_hermes=str(data.get("requires_hermes") or "").strip(),
config_spec=config_spec,
)
except Exception as exc:
logger.warning(
@@ -1899,24 +1748,6 @@ class PluginManager:
def _load_plugin(self, manifest: PluginManifest) -> None:
"""Import a plugin module and call its ``register(ctx)`` function."""
loaded = LoadedPlugin(manifest=manifest)
# requires_hermes gate — skip cleanly (no import, no traceback) when
# the running Hermes version doesn't satisfy the manifest spec.
if manifest.requires_hermes:
current = _running_hermes_version()
if not _version_satisfies(manifest.requires_hermes, current):
loaded.enabled = False
loaded.error = (
f"requires hermes {manifest.requires_hermes}, "
f"running {current}"
)
self._plugins[manifest.key or manifest.name] = loaded
logger.warning(
"Plugin '%s' skipped: %s",
manifest.key or manifest.name, loaded.error,
)
return
logger.debug(
"Loading plugin '%s' (source=%s, kind=%s, path=%s)",
manifest.key or manifest.name, manifest.source, manifest.kind, manifest.path,
+9 -668
View File
@@ -446,61 +446,19 @@ def _require_installed_plugin(name: str, plugins_dir: Path, console) -> Path:
# ---------------------------------------------------------------------------
def _raise_removed(removed) -> None:
"""Raise PluginOperationError describing a blocklisted plugin."""
detail = removed.reason or "no reason recorded"
if removed.date:
detail += f" (removed {removed.date})"
raise PluginOperationError(
f"Plugin '{removed.name}' was removed from the Hermes plugin "
f"catalog and is blocked from installation: {detail}"
)
def _install_plugin_core(
identifier: str,
*,
force: bool,
ref: Optional[str] = None,
skip_removed_check: bool = False,
) -> tuple[Path, dict, str]:
def _install_plugin_core(identifier: str, *, force: bool) -> tuple[Path, dict, str]:
"""Clone Git plugin into ``~/.hermes/plugins``.
``ref`` — optional git commit SHA (or tag) checked out after clone.
When given, the clone is full-depth (no ``--depth 1``) so any commit is
reachable.
Unless ``skip_removed_check`` is set, the identifier and the resolved
repo URL are checked against the plugin catalog's removed blocklist
(``plugin-catalog/removed.yaml``); a hit raises ``PluginOperationError``
with the recorded reason and date.
Returns ``(target_dir, installed_manifest, canonical_name)``.
Raises ``PluginOperationError`` on failure.
"""
import tempfile
if not skip_removed_check:
from hermes_cli.plugin_catalog import find_removed
# Check the raw identifier first (catches catalog names before URL
# resolution), then the resolved repo URL below.
removed = find_removed(identifier)
if removed is not None:
_raise_removed(removed)
try:
git_url, subdir = _resolve_git_url(identifier)
except ValueError as e:
raise PluginOperationError(str(e)) from e
if not skip_removed_check:
from hermes_cli.plugin_catalog import find_removed
removed = find_removed(git_url)
if removed is not None:
_raise_removed(removed)
plugins_dir = _plugins_dir()
with tempfile.TemporaryDirectory() as tmp:
@@ -510,15 +468,9 @@ def _install_plugin_core(
if not git_exe:
raise PluginOperationError("git is not installed or not in PATH.")
clone_cmd = [git_exe, "clone"]
if ref is None:
# Fast path — only the tip is needed.
clone_cmd += ["--depth", "1"]
clone_cmd += [git_url, str(tmp_clone)]
try:
result = subprocess.run(
clone_cmd,
[git_exe, "clone", "--depth", "1", git_url, str(tmp_clone)],
capture_output=True,
text=True,
timeout=60,
@@ -536,24 +488,6 @@ def _install_plugin_core(
err = (result.stderr or result.stdout or "").strip()
raise PluginOperationError(f"Git clone failed:\n{err}")
if ref is not None:
try:
checkout = subprocess.run(
[git_exe, "-C", str(tmp_clone), "checkout", ref],
capture_output=True,
text=True,
timeout=60,
)
except subprocess.TimeoutExpired as e:
raise PluginOperationError(
f"Git checkout of ref '{ref}' timed out after 60 seconds.",
) from e
if checkout.returncode != 0:
err = (checkout.stderr or checkout.stdout or "").strip()
raise PluginOperationError(
f"Git checkout of ref '{ref}' failed:\n{err}"
)
# Resolve the directory within the clone that holds the plugin.
if subdir:
tmp_target = _resolve_subdir_within(tmp_clone, subdir)
@@ -613,98 +547,12 @@ 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 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).
"""Install a plugin from a Git URL or owner/repo shorthand.
After install, prompt "Enable now? [y/N]" unless *enable* is provided
(True = auto-enable without prompting, False = install disabled).
@@ -713,51 +561,6 @@ 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:
@@ -779,16 +582,11 @@ 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():
@@ -836,13 +634,7 @@ def cmd_install(
def cmd_update(name: str) -> None:
"""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.
"""
"""Update an installed plugin by pulling latest from its git remote."""
from rich.console import Console
console = Console()
@@ -854,11 +646,6 @@ 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 "
@@ -886,55 +673,6 @@ 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
@@ -1384,44 +1122,6 @@ 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
@@ -1439,22 +1139,16 @@ def cmd_list(args: Any | None = None) -> None:
entries = _filter_plugin_entries(entries, args, enabled, disabled)
if getattr(args, "json", False):
payload = []
for name, version, description, source, _dir, key in entries:
row = {
payload = [
{
"name": name,
"status": _plugin_status(name, enabled, disabled, key=key),
"version": str(version),
"description": description,
"source": source,
}
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)
for name, version, description, source, _dir, key in entries
]
print(json.dumps(payload, indent=2))
return
@@ -1475,7 +1169,6 @@ 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":
@@ -1484,20 +1177,10 @@ def cmd_list(args: Any | None = None) -> None:
status = "[green]enabled[/green]"
else:
status = "[yellow]not enabled[/yellow]"
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]"
)
table.add_row(name, status, str(version), description, source)
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")
@@ -1505,337 +1188,6 @@ 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
# ---------------------------------------------------------------------------
@@ -2665,18 +2017,7 @@ 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
View File
@@ -42,51 +42,6 @@ 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"
-60
View File
@@ -1,60 +0,0 @@
# Hermes Plugin Catalog
Curated, Nous-approved Hermes plugins. Each YAML file in this directory
(except `removed.yaml`) is one catalog entry, discoverable via
`hermes plugins catalog` / `hermes plugins search` and installable with
`hermes plugins install <name>`.
## Admission policy
Presence in this directory **is** the trust signal. The rules that keep it
meaningful:
1. **Human-merged gate.** Entries are added *only* via a PR to the
`hermes-agent` repository, reviewed and merged by a maintainer. There is
no self-serve registry, no automated ingestion.
2. **Exact SHA pins are mandatory.** Every entry pins a full 40-character
commit SHA. Branches, tags, and short SHAs are rejected by the loader.
Installs clone the repository and check out exactly that commit.
3. **Pin maturity.** The pinned release should be **at least 2 weeks old**
at pin time, mirroring the supply-chain policy used for `optional-mcps/`
and pyproject dependencies. This gives the community time to notice a
compromised release before Hermes ships a pointer to it.
4. **SHA bumps are new PRs.** Updating an entry's pin is a new PR whose diff
(old SHA → new SHA) is re-reviewed like any other change — reviewers are
expected to look at the upstream commit range being adopted.
5. **Owner-or-major-contributor submissions only.** An entry may only be
submitted by the plugin repository's owner or a major contributor to it.
Drive-by submissions of third-party repos are declined.
6. **Declared capabilities must match reality.** The `capabilities:` block
(tools, hooks, middleware, env vars) must match what the plugin actually
registers at the pinned commit. Validation fails the entry otherwise —
undeclared capability creep is treated as a security issue.
## Entry schema
```yaml
name: example-plugin # [a-z0-9_-]{1,64}, the catalog key
repo: https://github.com/owner/repo # https:// only
sha: <40-hex commit sha> # mandatory exact pin
subdir: "" # optional path within the repo
description: One-line description.
maintainer: OwnerName
tier: official # official | community (default community)
requires_hermes: ">=0.19" # optional
docs_url: "" # optional
platforms: [] # optional, e.g. [linux, macos]; empty = all
capabilities:
provides_tools: []
provides_hooks: []
provides_middleware: []
requires_env: []
```
## removed.yaml — the blocklist
When an entry is pulled from the catalog for security or policy reasons, it
is recorded in `removed.yaml` with a reason and date. The installer refuses
to install anything matching a removed entry's name or repo URL, so a
malicious plugin cannot be re-installed from a stale identifier after
removal. Removals, like additions, land via reviewed PRs.
-14
View File
@@ -1,14 +0,0 @@
name: example-plugin
repo: https://github.com/NousResearch/hermes-example-plugins
sha: 38fe0fb53eff98d477f807432e965429e665ca33
subdir: ""
description: Reference example plugins for the Hermes plugin system.
maintainer: NousResearch
tier: official
docs_url: ""
platforms: []
capabilities:
provides_tools: []
provides_hooks: []
provides_middleware: []
requires_env: []
-6
View File
@@ -1,6 +0,0 @@
# Blocklist for plugins pulled from the catalog for security or policy
# reasons. The installer refuses to install anything whose name or repo URL
# matches an entry here (unless the caller explicitly bypasses the check).
# Each entry: {name, repo, reason, date}. Removals land via reviewed PRs,
# same as additions.
removed: []
+271
View File
@@ -0,0 +1,271 @@
#!/usr/bin/env python3
"""Standalone structural validator for plugin-catalog entry files.
Validates ``plugin-catalog/*.yaml`` catalog entries and
``plugin-catalog/removed.yaml`` against the catalog contract schema, using
only stdlib + PyYAML so the admission CI (and third-party repos) can run it
WITHOUT installing hermes-agent.
NOTE: this script intentionally duplicates the schema rules instead of
importing ``hermes_cli`` — the whole point is the no-install requirement for
cheap cross-repo CI use. The runtime twin of this schema lives in
``hermes_cli/plugin_catalog.py``; if the contract changes there, update the
rules here in lockstep.
Usage:
python3 scripts/validate_plugin_catalog.py plugin-catalog/
python3 scripts/validate_plugin_catalog.py entry.yaml removed.yaml
python3 scripts/validate_plugin_catalog.py --json plugin-catalog/
Exit codes: 0 = all files valid (warnings allowed), 1 = at least one error.
Human output is one ``<file>: ERROR: ...`` / ``<file>: warning: ...`` line
per finding; ``--json`` emits a machine-readable report on stdout instead.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path
try:
import yaml
except ImportError: # pragma: no cover - dependency guidance only
print(
"ERROR: PyYAML is required (pip install pyyaml)",
file=sys.stderr,
)
sys.exit(2)
NAME_RE = re.compile(r"^[a-z0-9_-]{1,64}$")
SHA_RE = re.compile(r"^[0-9a-f]{40}$")
TIERS = ("official", "community")
PLATFORMS = ("linux", "macos", "windows")
CAPABILITY_KEYS = (
"provides_tools",
"provides_hooks",
"provides_middleware",
"requires_env",
)
# Top-level keys the contract knows about. Unknown keys WARN (forward
# compatibility: newer catalogs must stay valid under older validators).
KNOWN_KEYS = {
"name",
"repo",
"sha",
"subdir",
"description",
"maintainer",
"tier",
"requires_hermes",
"docs_url",
"platforms",
"capabilities",
}
REQUIRED_KEYS = ("name", "repo", "sha", "description", "maintainer")
# One comparator clause of a requires_hermes spec, e.g. ">=0.19" or "!=1.2.3".
_COMPARATOR_RE = re.compile(r"^(>=|<=|==|!=|>|<)\s*\d+(\.\d+)*$")
def _is_nonempty_str(value: object) -> bool:
return isinstance(value, str) and value.strip() != ""
def _check_requires_hermes(spec: object, errors: list[str]) -> None:
if not isinstance(spec, str):
errors.append(f"requires_hermes must be a string, got {type(spec).__name__}")
return
if spec.strip() == "":
return # empty = no constraint
for clause in spec.split(","):
if not _COMPARATOR_RE.match(clause.strip()):
errors.append(
f"requires_hermes clause {clause.strip()!r} is not a valid "
"comparator spec (expected e.g. '>=0.19')"
)
def validate_entry(data: object) -> tuple[list[str], list[str]]:
"""Validate one catalog entry document. Returns (errors, warnings)."""
errors: list[str] = []
warnings: list[str] = []
if not isinstance(data, dict):
return ["top-level document must be a YAML mapping"], warnings
for key in REQUIRED_KEYS:
if key not in data:
errors.append(f"missing required key: {key}")
for key in sorted(set(data) - KNOWN_KEYS):
warnings.append(f"unknown top-level key {key!r} (ignored by this validator)")
name = data.get("name")
if "name" in data and (not isinstance(name, str) or not NAME_RE.match(name)):
errors.append(f"name {name!r} must match [a-z0-9_-]{{1,64}}")
repo = data.get("repo")
if "repo" in data and (
not isinstance(repo, str) or not repo.startswith("https://")
):
errors.append(f"repo {repo!r} must be an https:// URL")
sha = data.get("sha")
if "sha" in data and (not isinstance(sha, str) or not SHA_RE.match(sha)):
errors.append(f"sha {sha!r} must be exactly 40 lowercase hex characters")
for key in ("description", "maintainer"):
if key in data and not _is_nonempty_str(data[key]):
errors.append(f"{key} must be a non-empty string")
tier = data.get("tier", "community")
if tier not in TIERS:
errors.append(f"tier {tier!r} must be one of {list(TIERS)}")
if "requires_hermes" in data:
_check_requires_hermes(data["requires_hermes"], errors)
platforms = data.get("platforms", [])
if platforms is None:
platforms = []
if not isinstance(platforms, list):
errors.append("platforms must be a list")
else:
bad = [p for p in platforms if p not in PLATFORMS]
if bad:
errors.append(f"platforms {bad!r} not in allowed set {list(PLATFORMS)}")
caps = data.get("capabilities", {})
if caps is None:
caps = {}
if not isinstance(caps, dict):
errors.append("capabilities must be a mapping")
else:
for key in sorted(set(caps) - set(CAPABILITY_KEYS)):
warnings.append(f"unknown capabilities key {key!r}")
for key in CAPABILITY_KEYS:
if key not in caps:
continue
value = caps[key]
if not isinstance(value, list) or not all(
isinstance(item, str) for item in value
):
errors.append(f"capabilities.{key} must be a list of strings")
return errors, warnings
def validate_removed(data: object) -> tuple[list[str], list[str]]:
"""Validate the removed.yaml document. Returns (errors, warnings)."""
errors: list[str] = []
warnings: list[str] = []
if not isinstance(data, dict):
return ["top-level document must be a YAML mapping"], warnings
removed = data.get("removed")
if removed is None:
errors.append("missing required key: removed")
return errors, warnings
if not isinstance(removed, list):
errors.append("removed must be a list")
return errors, warnings
for i, item in enumerate(removed):
if not isinstance(item, dict):
errors.append(f"removed[{i}] must be a mapping")
continue
if not _is_nonempty_str(item.get("name")):
errors.append(f"removed[{i}] missing non-empty 'name'")
for key in ("repo", "reason", "date"):
if key in item and not isinstance(item[key], str):
errors.append(f"removed[{i}].{key} must be a string")
return errors, warnings
def validate_file(path: Path) -> tuple[list[str], list[str]]:
"""Validate one YAML file (dispatching on filename). Returns (errors, warnings)."""
try:
with open(path, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
except OSError as exc:
return [f"cannot read file: {exc}"], []
except yaml.YAMLError as exc:
return [f"invalid YAML: {exc}"], []
if path.name == "removed.yaml":
return validate_removed(data)
return validate_entry(data)
def collect_paths(args: list[str]) -> list[Path]:
paths: list[Path] = []
for arg in args:
p = Path(arg)
if p.is_dir():
paths.extend(sorted(p.glob("*.yaml")))
paths.extend(sorted(p.glob("*.yml")))
else:
paths.append(p)
return paths
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Standalone structural validator for plugin-catalog entry files."
)
parser.add_argument(
"paths",
nargs="+",
help="catalog entry files, removed.yaml, or a directory of them",
)
parser.add_argument(
"--json",
action="store_true",
help="emit a machine-readable JSON report on stdout",
)
opts = parser.parse_args(argv)
files = collect_paths(opts.paths)
if not files:
print("ERROR: no YAML files found", file=sys.stderr)
return 1
report = []
any_errors = False
for path in files:
errors, warnings = validate_file(path)
any_errors = any_errors or bool(errors)
report.append(
{
"path": str(path),
"ok": not errors,
"errors": errors,
"warnings": warnings,
}
)
if opts.json:
print(json.dumps({"ok": not any_errors, "files": report}, indent=2))
else:
for entry in report:
for err in entry["errors"]:
print(f"{entry['path']}: ERROR: {err}")
for warn in entry["warnings"]:
print(f"{entry['path']}: warning: {warn}")
checked = len(report)
bad = sum(1 for e in report if not e["ok"])
if any_errors:
print(f"FAIL: {bad}/{checked} file(s) invalid")
else:
print(f"OK: {checked} file(s) valid")
return 1 if any_errors else 0
if __name__ == "__main__":
sys.exit(main())
-594
View File
@@ -1,594 +0,0 @@
"""Tests for the Hermes plugin catalog (hermes_cli.plugin_catalog) and the
catalog-driven install/manifest extensions in plugins_cmd.py / plugins.py."""
from __future__ import annotations
import os
import subprocess
from pathlib import Path
import pytest
import yaml
from hermes_cli.plugin_catalog import (
CATALOG_TIERS,
PluginCatalogEntry,
RemovedEntry,
entry_capability_summary,
find_removed,
get_catalog_dir,
get_catalog_entry,
load_catalog,
load_removed_list,
search_catalog,
)
VALID_SHA = "38fe0fb53eff98d477f807432e965429e665ca33"
# ── Helpers ────────────────────────────────────────────────────────────────
def _write_entry(catalog_dir: Path, name: str, **overrides) -> Path:
"""Write a minimal valid catalog entry yaml, applying overrides."""
data = {
"name": name,
"repo": f"https://github.com/example/{name}",
"sha": VALID_SHA,
"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
@pytest.fixture()
def catalog_dir(tmp_path, monkeypatch):
d = tmp_path / "catalog"
d.mkdir()
monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(d))
return d
# ── get_catalog_dir ────────────────────────────────────────────────────────
class TestGetCatalogDir:
def test_env_override_wins(self, catalog_dir):
assert get_catalog_dir() == catalog_dir
def test_default_is_repo_plugin_catalog(self, monkeypatch):
monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False)
d = get_catalog_dir()
assert d.name == "plugin-catalog"
# ── load_catalog ───────────────────────────────────────────────────────────
class TestLoadCatalog:
def test_valid_entry_parses(self, catalog_dir):
_write_entry(
catalog_dir,
"my-plugin",
tier="official",
requires_hermes=">=0.19",
subdir="plugins/my-plugin",
docs_url="https://example.com/docs",
platforms=["linux"],
capabilities={
"provides_tools": ["my_tool"],
"provides_hooks": ["on_start"],
"provides_middleware": ["llm_request"],
"requires_env": ["MY_API_KEY"],
},
)
entries = load_catalog()
assert len(entries) == 1
e = entries[0]
assert isinstance(e, PluginCatalogEntry)
assert e.name == "my-plugin"
assert e.repo == "https://github.com/example/my-plugin"
assert e.sha == VALID_SHA
assert e.tier == "official"
assert e.requires_hermes == ">=0.19"
assert e.subdir == "plugins/my-plugin"
assert e.docs_url == "https://example.com/docs"
assert e.platforms == ["linux"]
assert e.capabilities.provides_tools == ["my_tool"]
assert e.capabilities.provides_hooks == ["on_start"]
assert e.capabilities.provides_middleware == ["llm_request"]
assert e.capabilities.requires_env == ["MY_API_KEY"]
def test_tier_defaults_to_community(self, catalog_dir):
_write_entry(catalog_dir, "no-tier")
(entry,) = load_catalog()
assert entry.tier == "community"
assert entry.tier in CATALOG_TIERS
def test_bad_sha_rejected(self, catalog_dir, caplog):
_write_entry(catalog_dir, "bad-sha", sha="main")
_write_entry(catalog_dir, "short-sha", sha="38fe0fb")
_write_entry(catalog_dir, "good", sha=VALID_SHA)
with caplog.at_level("WARNING"):
entries = load_catalog()
assert [e.name for e in entries] == ["good"]
def test_bad_name_rejected(self, catalog_dir, caplog):
_write_entry(catalog_dir, "BadName")
_write_entry(catalog_dir, "has spaces")
with caplog.at_level("WARNING"):
entries = load_catalog()
assert entries == []
def test_non_https_repo_rejected(self, catalog_dir, caplog):
_write_entry(catalog_dir, "sshrepo", repo="git@github.com:x/y.git")
with caplog.at_level("WARNING"):
entries = load_catalog()
assert entries == []
def test_invalid_tier_rejected(self, catalog_dir, caplog):
_write_entry(catalog_dir, "weird-tier", tier="platinum")
with caplog.at_level("WARNING"):
entries = load_catalog()
assert entries == []
def test_removed_yaml_is_not_an_entry(self, catalog_dir):
_write_entry(catalog_dir, "real-entry")
_write_removed(catalog_dir, [])
entries = load_catalog()
assert [e.name for e in entries] == ["real-entry"]
def test_unparseable_yaml_skipped_without_raising(self, catalog_dir, caplog):
(catalog_dir / "broken.yaml").write_text(
"name: [unclosed", encoding="utf-8"
)
_write_entry(catalog_dir, "ok-entry")
with caplog.at_level("WARNING"):
entries = load_catalog()
assert [e.name for e in entries] == ["ok-entry"]
def test_missing_dir_returns_empty(self, tmp_path, monkeypatch):
monkeypatch.setenv(
"HERMES_PLUGIN_CATALOG_DIR", str(tmp_path / "does-not-exist")
)
assert load_catalog() == []
# ── get_catalog_entry / search_catalog ─────────────────────────────────────
class TestLookupAndSearch:
def test_get_catalog_entry_by_name(self, catalog_dir):
_write_entry(catalog_dir, "alpha")
_write_entry(catalog_dir, "beta")
entry = get_catalog_entry("beta")
assert entry is not None and entry.name == "beta"
assert get_catalog_entry("nope") is None
def test_search_matches_name_case_insensitive(self, catalog_dir):
_write_entry(catalog_dir, "weather-tools")
_write_entry(catalog_dir, "other")
results = search_catalog("WEATHER")
assert [e.name for e in results] == ["weather-tools"]
def test_search_matches_description(self, catalog_dir):
_write_entry(catalog_dir, "abc", description="Fetches Stock Quotes.")
results = search_catalog("stock")
assert [e.name for e in results] == ["abc"]
def test_search_matches_declared_tools(self, catalog_dir):
_write_entry(
catalog_dir,
"toolful",
capabilities={"provides_tools": ["get_forecast"]},
)
_write_entry(catalog_dir, "toolless")
results = search_catalog("Forecast")
assert [e.name for e in results] == ["toolful"]
def test_empty_query_returns_all(self, catalog_dir):
_write_entry(catalog_dir, "one")
_write_entry(catalog_dir, "two")
assert len(search_catalog("")) == 2
# ── removed list ───────────────────────────────────────────────────────────
class TestRemovedList:
def test_load_removed_list(self, catalog_dir):
_write_removed(
catalog_dir,
[
{
"name": "evil-plugin",
"repo": "https://github.com/evil/evil-plugin",
"reason": "Exfiltrated env vars",
"date": "2026-07-02",
}
],
)
removed = load_removed_list()
assert len(removed) == 1
r = removed[0]
assert isinstance(r, RemovedEntry)
assert r.name == "evil-plugin"
assert r.reason == "Exfiltrated env vars"
assert r.date == "2026-07-02"
def test_missing_removed_yaml_returns_empty(self, catalog_dir):
assert load_removed_list() == []
assert find_removed("anything") is None
def test_find_removed_by_name(self, catalog_dir):
_write_removed(catalog_dir, [{"name": "evil-plugin", "reason": "bad"}])
hit = find_removed("evil-plugin")
assert hit is not None and hit.reason == "bad"
def test_find_removed_by_repo_url_with_and_without_git_suffix(
self, catalog_dir
):
_write_removed(
catalog_dir,
[
{
"name": "evil-plugin",
"repo": "https://github.com/evil/evil-plugin",
"reason": "bad",
}
],
)
assert find_removed("https://github.com/evil/evil-plugin") is not None
assert find_removed("https://github.com/evil/evil-plugin.git") is not None
assert find_removed("https://github.com/good/fine.git") is None
# ── entry_capability_summary ───────────────────────────────────────────────
class TestCapabilitySummary:
def test_summary_contains_declared_capabilities(self):
entry = PluginCatalogEntry(
name="cap-plugin",
repo="https://github.com/example/cap-plugin",
sha=VALID_SHA,
description="Does capable things.",
maintainer="Example",
)
entry.capabilities.provides_tools = ["tool_a", "tool_b"]
entry.capabilities.provides_hooks = ["session_start"]
entry.capabilities.requires_env = ["CAP_API_KEY"]
summary = entry_capability_summary(entry)
assert "tool_a" in summary
assert "tool_b" in summary
assert "session_start" in summary
assert "CAP_API_KEY" in summary
def test_summary_for_empty_capabilities_mentions_none(self):
entry = PluginCatalogEntry(
name="plain",
repo="https://github.com/example/plain",
sha=VALID_SHA,
description="Plain.",
maintainer="Example",
)
summary = entry_capability_summary(entry)
assert summary # non-empty human text
# ── shipped catalog seed ───────────────────────────────────────────────────
class TestShippedCatalog:
def test_shipped_catalog_entries_are_valid(self, monkeypatch):
"""Every yaml shipped in <repo>/plugin-catalog must load cleanly."""
monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False)
shipped = get_catalog_dir()
yaml_files = [
p for p in shipped.glob("*.yaml") if p.name != "removed.yaml"
]
entries = load_catalog()
assert len(entries) == len(yaml_files)
# removed.yaml must exist and parse
assert (shipped / "removed.yaml").exists()
load_removed_list()
# ── _version_satisfies ─────────────────────────────────────────────────────
class TestVersionSatisfies:
@pytest.fixture(autouse=True)
def _import(self):
from hermes_cli.plugins import _version_satisfies
self.satisfies = _version_satisfies
def test_ge(self):
assert self.satisfies(">=0.19", "0.19.0") is True
assert self.satisfies(">=0.19", "0.20.1") is True
assert self.satisfies(">=0.19", "0.18.2") is False
def test_gt_lt_le(self):
assert self.satisfies(">0.19", "0.19.1") is True
assert self.satisfies(">0.19", "0.19.0") is False
assert self.satisfies("<1.0", "0.19.0") is True
assert self.satisfies("<=0.19.0", "0.19.0") is True
def test_eq_ne(self):
assert self.satisfies("==0.19.0", "0.19.0") is True
assert self.satisfies("==0.19.0", "0.19.1") is False
assert self.satisfies("!=0.19.0", "0.19.1") is True
assert self.satisfies("!=0.19.0", "0.19.0") is False
def test_comma_separated_all_must_hold(self):
assert self.satisfies(">=0.10, <1.0", "0.19.0") is True
assert self.satisfies(">=0.10, <0.15", "0.19.0") is False
def test_bare_version_treated_as_ge(self):
assert self.satisfies("0.10", "0.19.0") is True
assert self.satisfies("999", "0.19.0") is False
def test_empty_spec_is_satisfied(self):
assert self.satisfies("", "0.19.0") is True
def test_non_numeric_segments_fall_back_permissive(self):
assert self.satisfies(">=abc.def", "0.19.0") is True
assert self.satisfies(">=0.19", "unknown") is True
# ── requires_hermes manifest gate ──────────────────────────────────────────
def _make_plugin(base: Path, name: str, *, manifest_extra: dict | None = None,
register_body: str = "pass", enable: bool = True) -> Path:
"""Create a plugin dir under <HERMES_HOME>/plugins and opt it in."""
plugin_dir = base / name
plugin_dir.mkdir(parents=True, exist_ok=True)
manifest = {"name": name, "version": "0.1.0", "description": name}
if manifest_extra:
manifest.update(manifest_extra)
(plugin_dir / "plugin.yaml").write_text(yaml.safe_dump(manifest))
(plugin_dir / "__init__.py").write_text(
f"def register(ctx):\n {register_body}\n"
)
if enable:
hermes_home = Path(os.environ["HERMES_HOME"])
cfg_path = hermes_home / "config.yaml"
cfg: dict = {}
if cfg_path.exists():
cfg = yaml.safe_load(cfg_path.read_text()) or {}
cfg.setdefault("plugins", {}).setdefault("enabled", []).append(name)
cfg_path.write_text(yaml.safe_dump(cfg))
return plugin_dir
class TestRequiresHermesGate:
def test_unsatisfied_requires_hermes_skips_load(self, monkeypatch):
from hermes_cli.plugins import PluginManager
hermes_home = Path(os.environ["HERMES_HOME"])
plugins_dir = hermes_home / "plugins"
_make_plugin(
plugins_dir, "future_plugin",
manifest_extra={"requires_hermes": ">=999.0"},
)
mgr = PluginManager()
mgr.discover_and_load()
loaded = mgr._plugins["future_plugin"]
assert loaded.enabled is False
assert loaded.error is not None
assert "requires hermes" in loaded.error
assert ">=999.0" in loaded.error
assert loaded.module is None # register() never ran
def test_satisfied_requires_hermes_loads_normally(self, monkeypatch):
from hermes_cli.plugins import PluginManager
hermes_home = Path(os.environ["HERMES_HOME"])
plugins_dir = hermes_home / "plugins"
_make_plugin(
plugins_dir, "old_ok_plugin",
manifest_extra={"requires_hermes": ">=0.1"},
)
mgr = PluginManager()
mgr.discover_and_load()
loaded = mgr._plugins["old_ok_plugin"]
assert loaded.enabled is True
assert loaded.error is None
def test_requires_hermes_parsed_onto_manifest(self):
from hermes_cli.plugins import PluginManager
hermes_home = Path(os.environ["HERMES_HOME"])
plugins_dir = hermes_home / "plugins"
_make_plugin(
plugins_dir, "spec_plugin",
manifest_extra={"requires_hermes": ">=0.19"},
enable=False,
)
mgr = PluginManager()
mgr.discover_and_load()
assert mgr._plugins["spec_plugin"].manifest.requires_hermes == ">=0.19"
# ── config: spec parsing + ctx.plugin_config ───────────────────────────────
class TestPluginConfig:
def test_config_spec_parsed_onto_manifest(self):
from hermes_cli.plugins import PluginManager
hermes_home = Path(os.environ["HERMES_HOME"])
plugins_dir = hermes_home / "plugins"
spec = [
{"key": "api_url", "prompt": "API URL", "type": "str",
"default": "https://api.example.com", "secret": False},
{"key": "token", "prompt": "Token", "type": "str", "secret": True},
]
_make_plugin(
plugins_dir, "cfg_plugin",
manifest_extra={"config": spec},
enable=False,
)
mgr = PluginManager()
mgr.discover_and_load()
manifest = mgr._plugins["cfg_plugin"].manifest
assert isinstance(manifest.config_spec, list)
assert manifest.config_spec[0]["key"] == "api_url"
assert manifest.config_spec[1]["secret"] is True
def test_plugin_config_merges_defaults_under_config_entries(self):
from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager
hermes_home = Path(os.environ["HERMES_HOME"])
cfg_path = hermes_home / "config.yaml"
cfg_path.write_text(yaml.safe_dump({
"plugins": {"entries": {"merge_plugin": {"api_url": "https://override"}}}
}))
manifest = PluginManifest(
name="merge_plugin",
key="merge_plugin",
config_spec=[
{"key": "api_url", "default": "https://default"},
{"key": "retries", "type": "int", "default": 3},
],
)
ctx = PluginContext(manifest, PluginManager())
cfg = ctx.plugin_config
assert cfg["api_url"] == "https://override" # config.yaml wins
assert cfg["retries"] == 3 # default fills the gap
def test_plugin_config_empty_without_spec_or_entries(self):
from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager
manifest = PluginManifest(name="bare_plugin", key="bare_plugin")
ctx = PluginContext(manifest, PluginManager())
assert ctx.plugin_config == {}
# ── _install_plugin_core: ref checkout + removed blocklist ────────────────
def _make_git_repo(tmp_path: Path) -> tuple[Path, str, str]:
"""Create a local git repo with two commits; return (path, sha1, sha2)."""
repo = tmp_path / "src-repo"
repo.mkdir()
def git(*args):
subprocess.run(
["git", *args], cwd=repo, check=True, capture_output=True, text=True,
env={**os.environ,
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"},
)
git("init", "-b", "main")
(repo / "plugin.yaml").write_text(
yaml.safe_dump({"name": "refplugin", "version": "1"})
)
(repo / "__init__.py").write_text("def register(ctx):\n pass\n")
(repo / "marker.txt").write_text("first\n")
git("add", "-A")
git("commit", "-m", "first")
sha1 = subprocess.run(
["git", "rev-parse", "HEAD"], cwd=repo, check=True,
capture_output=True, text=True,
).stdout.strip()
(repo / "marker.txt").write_text("second\n")
git("add", "-A")
git("commit", "-m", "second")
sha2 = subprocess.run(
["git", "rev-parse", "HEAD"], cwd=repo, check=True,
capture_output=True, text=True,
).stdout.strip()
return repo, sha1, sha2
class TestInstallPluginCore:
def test_ref_checkout_installs_pinned_commit(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import _install_plugin_core
repo, sha1, _sha2 = _make_git_repo(tmp_path)
target, manifest, name = _install_plugin_core(
f"file://{repo}", force=False, ref=sha1
)
assert name == "refplugin"
assert (target / "marker.txt").read_text() == "first\n"
def test_default_install_gets_head(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import _install_plugin_core
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
target, _manifest, _name = _install_plugin_core(
f"file://{repo}", force=False
)
assert (target / "marker.txt").read_text() == "second\n"
def test_bad_ref_raises(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
with pytest.raises(PluginOperationError):
_install_plugin_core(
f"file://{repo}", force=False,
ref="0000000000000000000000000000000000000000",
)
def test_removed_repo_blocked(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
_write_removed(
catalog_dir,
[{
"name": "refplugin",
"repo": f"file://{repo}",
"reason": "exfiltrated env vars",
"date": "2026-07-02",
}],
)
with pytest.raises(PluginOperationError, match="exfiltrated env vars"):
_install_plugin_core(f"file://{repo}", force=False)
def test_removed_identifier_blocked_by_name(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
_write_removed(
catalog_dir,
[{"name": "evil-plugin", "reason": "malware", "date": "2026-01-01"}],
)
with pytest.raises(PluginOperationError, match="malware"):
_install_plugin_core("evil-plugin", force=False)
def test_skip_removed_check_bypasses_block(self, tmp_path, catalog_dir):
from hermes_cli.plugins_cmd import _install_plugin_core
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
_write_removed(
catalog_dir,
[{
"name": "refplugin",
"repo": f"file://{repo}",
"reason": "bad",
"date": "2026-07-02",
}],
)
target, _manifest, name = _install_plugin_core(
f"file://{repo}", force=False, skip_removed_check=True
)
assert name == "refplugin"
assert target.exists()
-214
View File
@@ -1,214 +0,0 @@
"""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
@@ -1,626 +0,0 @@
"""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]
@@ -0,0 +1,273 @@
"""Behavior tests for scripts/validate_plugin_catalog.py.
The script is the no-install structural validator used by the plugin-catalog
admission CI: it must run with only stdlib + pyyaml, take file paths or a
directory, exit 0/1, and support --json machine output. These tests exercise
the CLI contract via subprocess (the same way CI invokes it).
"""
import json
import subprocess
import sys
from pathlib import Path
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
SCRIPT = REPO_ROOT / "scripts" / "validate_plugin_catalog.py"
VALID_ENTRY = {
"name": "example-plugin",
"repo": "https://github.com/NousResearch/hermes-example-plugins",
"sha": "38fe0fb53eff98d477f807432e965429e665ca33",
"subdir": "",
"description": "One-line description.",
"maintainer": "NousResearch",
"tier": "official",
"requires_hermes": ">=0.19",
"docs_url": "",
"platforms": [],
"capabilities": {
"provides_tools": ["example_tool"],
"provides_hooks": [],
"provides_middleware": [],
"requires_env": [],
},
}
def write_entry(tmp_path: Path, data: dict, filename: str | None = None) -> Path:
name = filename or f"{data.get('name', 'entry')}.yaml"
path = tmp_path / name
path.write_text(yaml.safe_dump(data), encoding="utf-8")
return path
def run_validator(*args: str) -> subprocess.CompletedProcess:
return subprocess.run(
[sys.executable, str(SCRIPT), *args],
capture_output=True,
text=True,
)
# ── valid input ────────────────────────────────────────────────────────
def test_valid_entry_passes(tmp_path):
path = write_entry(tmp_path, VALID_ENTRY)
result = run_validator(str(path))
assert result.returncode == 0, result.stdout + result.stderr
def test_valid_entry_without_optional_fields_passes(tmp_path):
entry = {
"name": "minimal-plugin",
"repo": "https://github.com/example/minimal",
"sha": "a" * 40,
"description": "Minimal.",
"maintainer": "someone",
}
path = write_entry(tmp_path, entry)
result = run_validator(str(path))
assert result.returncode == 0, result.stdout + result.stderr
# ── each malformed field fails with a pointed error ────────────────────
def _expect_error(tmp_path, mutation: dict, expected_substring: str, drop: str = ""):
entry = {**VALID_ENTRY, **mutation}
if drop:
entry.pop(drop, None)
path = write_entry(tmp_path, entry, filename="entry.yaml")
result = run_validator(str(path))
combined = result.stdout + result.stderr
assert result.returncode == 1, combined
assert expected_substring in combined, combined
assert "entry.yaml" in combined, combined
def test_bad_name_fails(tmp_path):
_expect_error(tmp_path, {"name": "Bad Name!"}, "name")
def test_name_too_long_fails(tmp_path):
_expect_error(tmp_path, {"name": "x" * 65}, "name")
def test_non_https_repo_fails(tmp_path):
_expect_error(tmp_path, {"repo": "git@github.com:evil/x.git"}, "repo")
def test_short_sha_fails(tmp_path):
_expect_error(tmp_path, {"sha": "abc123"}, "sha")
def test_non_hex_sha_fails(tmp_path):
_expect_error(tmp_path, {"sha": "z" * 40}, "sha")
def test_bad_tier_fails(tmp_path):
_expect_error(tmp_path, {"tier": "platinum"}, "tier")
def test_empty_description_fails(tmp_path):
_expect_error(tmp_path, {"description": ""}, "description")
def test_empty_maintainer_fails(tmp_path):
_expect_error(tmp_path, {"maintainer": ""}, "maintainer")
def test_missing_required_field_fails(tmp_path):
_expect_error(tmp_path, {}, "sha", drop="sha")
def test_capabilities_value_not_a_list_fails(tmp_path):
_expect_error(
tmp_path,
{"capabilities": {"provides_tools": "not-a-list"}},
"provides_tools",
)
def test_capabilities_list_of_non_strings_fails(tmp_path):
_expect_error(
tmp_path,
{"capabilities": {"requires_env": [1, 2]}},
"requires_env",
)
def test_bad_requires_hermes_spec_fails(tmp_path):
_expect_error(tmp_path, {"requires_hermes": "banana"}, "requires_hermes")
def test_comma_separated_requires_hermes_passes(tmp_path):
entry = {**VALID_ENTRY, "requires_hermes": ">=0.19, <2.0"}
path = write_entry(tmp_path, entry)
result = run_validator(str(path))
assert result.returncode == 0, result.stdout + result.stderr
def test_unknown_platform_fails(tmp_path):
_expect_error(tmp_path, {"platforms": ["linux", "amiga"]}, "platforms")
def test_entry_not_a_mapping_fails(tmp_path):
path = tmp_path / "entry.yaml"
path.write_text("- just\n- a\n- list\n", encoding="utf-8")
result = run_validator(str(path))
assert result.returncode == 1
assert "mapping" in (result.stdout + result.stderr)
# ── unknown top-level keys warn but do not fail ────────────────────────
def test_unknown_key_warns_but_passes(tmp_path):
entry = {**VALID_ENTRY, "future_field": "hello"}
path = write_entry(tmp_path, entry)
result = run_validator(str(path))
combined = result.stdout + result.stderr
assert result.returncode == 0, combined
assert "future_field" in combined
assert "warning" in combined.lower()
# ── removed.yaml shape ─────────────────────────────────────────────────
def test_valid_removed_yaml_passes(tmp_path):
path = tmp_path / "removed.yaml"
path.write_text(
yaml.safe_dump(
{
"removed": [
{
"name": "some-plugin",
"repo": "https://github.com/evil/some-plugin",
"reason": "Exfiltrated env vars",
"date": "2026-07-02",
}
]
}
),
encoding="utf-8",
)
result = run_validator(str(path))
assert result.returncode == 0, result.stdout + result.stderr
def test_removed_yaml_not_a_list_fails(tmp_path):
path = tmp_path / "removed.yaml"
path.write_text(yaml.safe_dump({"removed": "nope"}), encoding="utf-8")
result = run_validator(str(path))
assert result.returncode == 1
assert "removed" in (result.stdout + result.stderr)
def test_removed_item_missing_name_fails(tmp_path):
path = tmp_path / "removed.yaml"
path.write_text(
yaml.safe_dump({"removed": [{"reason": "bad", "date": "2026-01-01"}]}),
encoding="utf-8",
)
result = run_validator(str(path))
assert result.returncode == 1
assert "name" in (result.stdout + result.stderr)
# ── --json machine output ──────────────────────────────────────────────
def test_json_output_shape_on_failure(tmp_path):
bad = write_entry(tmp_path, {**VALID_ENTRY, "sha": "short"}, filename="bad.yaml")
result = run_validator("--json", str(bad))
assert result.returncode == 1
payload = json.loads(result.stdout)
assert payload["ok"] is False
assert isinstance(payload["files"], list)
entry = next(f for f in payload["files"] if f["path"].endswith("bad.yaml"))
assert entry["ok"] is False
assert any("sha" in e for e in entry["errors"])
def test_json_output_shape_on_success_with_warning(tmp_path):
good = write_entry(tmp_path, {**VALID_ENTRY, "future_field": 1})
result = run_validator("--json", str(good))
assert result.returncode == 0
payload = json.loads(result.stdout)
assert payload["ok"] is True
(entry,) = payload["files"]
assert entry["ok"] is True
assert entry["errors"] == []
assert any("future_field" in w for w in entry["warnings"])
# ── directory mode ─────────────────────────────────────────────────────
def test_directory_mode_validates_all_entries_and_removed(tmp_path):
write_entry(tmp_path, VALID_ENTRY)
write_entry(tmp_path, {**VALID_ENTRY, "name": "bad-one", "sha": "nope"})
(tmp_path / "removed.yaml").write_text(
yaml.safe_dump({"removed": [{"name": "gone", "reason": "test"}]}),
encoding="utf-8",
)
result = run_validator(str(tmp_path))
combined = result.stdout + result.stderr
assert result.returncode == 1
assert "bad-one.yaml" in combined
# the valid entry and removed.yaml must not produce errors
assert combined.count("ERROR") == combined.count("bad-one.yaml: ERROR")
def test_directory_mode_all_valid_exits_zero(tmp_path):
write_entry(tmp_path, VALID_ENTRY)
(tmp_path / "removed.yaml").write_text(
yaml.safe_dump({"removed": []}), encoding="utf-8"
)
result = run_validator(str(tmp_path))
assert result.returncode == 0, result.stdout + result.stderr