feat(plugins): catalog is the sole discovery system — re-port onto main's layout

- hermes_cli/plugins_cmd_catalog.py: new sibling owning resolution, the
  .hermes-catalog.json provenance sidecar, search/info/validate, re-pin on
  update, and the dashboard/TUI payload builders. plugins_cmd.py only
  gains the hooks (cmd_install catalog branch, cmd_update / dashboard
  update re-pin, dashboard_install_plugin catalog_name + kill list,
  dispatch entries); the community index (plugin_index.py) is gone.
- hermes_cli/plugin_catalog.py: catalog_dir parameter replaces the
  test-only HERMES_PLUGIN_CATALOG_DIR env var; live refresh reads ONE
  published document (/docs/api/plugin-catalog.json, 6h cache, in-tree
  fallback) instead of the unauthenticated GitHub contents API (60 req/h,
  1 request per entry); in-tree and live removals are unioned so a stale
  cache can never un-block.
- Catalog route lives in web_routers/dashboard_ui.py (the facade is off
  limits); _plugin_runtime_status shared from web_server_dashboard.py;
  hub rows carry removed_reason. TUI plugins.manage gains catalog_name
  install, catalog row fields and an update action.
- plugin_validate: the probe context honours ctx.get_config defaults
  (real plugins do int(ctx.get_config("timeout", 180)) in register()).
- Installed-state merge matches through the sidecar's catalog_name
  first — catalog names rarely equal manifest names.
- extract-plugins.py emits plugin-catalog.json; deploy-site triggers on
  plugin-catalog/** so entry merges republish it.
This commit is contained in:
Teknium
2026-09-09 04:38:01 -07:00
parent 4e312cf22d
commit 41b4555ed9
15 changed files with 711 additions and 436 deletions

View File

@@ -9,6 +9,9 @@ on:
- 'website/**'
- 'skills/**'
- 'optional-skills/**'
# Catalog entry/removal merges must republish /docs/api/plugin-catalog.json —
# installed clients fetch it for live catalog refresh.
- 'plugin-catalog/**'
- '.github/workflows/deploy-site.yml'
workflow_dispatch:
inputs:

1
.gitignore vendored
View File

@@ -136,6 +136,7 @@ website/static/api/skills-meta.json
# plugins.json + plugins-meta.json are build artifacts emitted by
# website/scripts/extract-plugins.py during prebuild (Plugin Catalog page).
website/static/api/plugins.json
website/static/api/plugin-catalog.json
website/static/api/plugins-meta.json
# automation-blueprints-index.json is a build artifact emitted by
# website/scripts/extract-automation-blueprints.py during prebuild.

View File

@@ -1,52 +1,47 @@
"""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.
Mirrors the ``optional-mcps/`` MCP-catalog pattern: one YAML file per entry under the in-tree
``plugin-catalog/`` directory, pinned to an exact 40-character commit SHA. Presence in the directory IS
the human-merged approval gate; SHA bumps are new, re-reviewed PRs; ``removed.yaml`` is the kill list
(installs of a removed name/repo are refused with the recorded reason). Full policy:
``plugin-catalog/README.md``.
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.
Live refresh: the docs build publishes the same data as ONE JSON document
(``website/scripts/extract-plugins.py`` → ``/docs/api/plugin-catalog.json``, like the skills index), so
an installed Hermes sees new entries and removals without updating. Any fetch failure falls back to the
in-tree copy silently.
"""
from __future__ import annotations
import json
import logging
import os
import re
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, List, Optional
from typing import Any, Dict, List, Optional
import yaml
logger = logging.getLogger(__name__)
CATALOG_TIERS = ("official", "community")
LIVE_CATALOG_URL = "https://hermes-agent.nousresearch.com/docs/api/plugin-catalog.json"
LIVE_CATALOG_TTL_SECONDS = 6 * 60 * 60
_REQUEST_TIMEOUT = 5.0
_MAX_LIVE_BYTES = 2 * 1024 * 1024
_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
date: str = ""
@dataclass
@@ -61,354 +56,230 @@ class CatalogCapabilities:
class PluginCatalogEntry:
name: str # catalog key, [a-z0-9_-]{1,64}
repo: str # https:// git URL
sha: str # 40-hex pinned commit — MANDATORY, validated
sha: str # 40-hex pinned commit — mandatory
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
tier: str = "community"
requires_hermes: str = ""
subdir: str = ""
docs_url: str = ""
platforms: List[str] = field(default_factory=list) # empty = all OSes
capabilities: CatalogCapabilities = field(default_factory=CatalogCapabilities)
@property
def install_identifier(self) -> str:
"""``_install_plugin_core`` identifier (``repo#subdir`` for monorepo entries)."""
return f"{self.repo}#{self.subdir}" if self.subdir else self.repo
# ─── Directory resolution ────────────────────────────────────────────────────
def to_dict(self) -> Dict[str, Any]:
caps = self.capabilities
return {
"name": self.name, "repo": self.repo, "sha": self.sha, "description": self.description,
"maintainer": self.maintainer, "tier": self.tier, "requires_hermes": self.requires_hermes,
"subdir": self.subdir, "docs_url": self.docs_url, "platforms": list(self.platforms),
"capabilities": {
"provides_tools": list(caps.provides_tools), "provides_hooks": list(caps.provides_hooks),
"provides_middleware": list(caps.provides_middleware), "requires_env": list(caps.requires_env),
},
}
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)
"""The ``plugin-catalog/`` directory shipped with this checkout."""
return Path(__file__).resolve().parent.parent / "plugin-catalog"
# ─── Loading / validation ────────────────────────────────────────────────────
# ── Parsing ──────────────────────────────────────────────────────────────────
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))]
return [str(x) for x in raw if isinstance(x, (str, int, float))] if isinstance(raw, list) else []
def _parse_entry(path: Path) -> Optional[PluginCatalogEntry]:
"""Parse and validate one catalog YAML file.
def entry_from_mapping(data: Any, label: str) -> Optional[PluginCatalogEntry]:
"""Validate one entry mapping (YAML file or live-JSON element); ``None`` + warning on any failure."""
if not isinstance(data, dict):
logger.warning("Plugin catalog: %s: entry must be a mapping", label)
return None
name = str(data.get("name") or "")
repo = str(data.get("repo") or "")
sha = str(data.get("sha") or "").strip().lower()
tier = str(data.get("tier") or "community")
problem = (
f"invalid name {name!r} (must match [a-z0-9_-]{{1,64}})" if not _NAME_RE.match(name)
else f"repo must be an https:// URL (got {repo!r})" if not repo.startswith("https://")
else f"sha must be a full 40-character hex commit SHA (got {data.get('sha')!r})" if not _SHA_RE.match(sha)
else f"tier must be one of {'/'.join(CATALOG_TIERS)} (got {tier!r})" if tier not in CATALOG_TIERS
else None)
if problem:
logger.warning("Plugin catalog: %s: %s", label, problem)
return None
caps_raw = data.get("capabilities")
caps: Dict[str, Any] = caps_raw if isinstance(caps_raw, dict) else {}
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=CatalogCapabilities(
provides_tools=_str_list(caps.get("provides_tools")), provides_hooks=_str_list(caps.get("provides_hooks")),
provides_middleware=_str_list(caps.get("provides_middleware")),
requires_env=_str_list(caps.get("requires_env"))),
)
Returns ``None`` (after logging a warning) on any validation failure —
the loader never raises for a bad entry.
"""
def _read_yaml(path: Path) -> Any:
try:
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
return 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 _removed_from_list(raw_list: Any) -> List[RemovedEntry]:
if not isinstance(raw_list, list):
return []
return [
RemovedEntry(name=str(r["name"]), repo=str(r.get("repo") or ""), reason=str(r.get("reason") or ""),
date=str(r.get("date") or ""))
for r in raw_list if isinstance(r, dict) and r.get("name")]
def load_catalog() -> List[PluginCatalogEntry]:
"""Return all valid catalog entries, sorted by name.
# ── In-tree catalog ──────────────────────────────────────────────────────────
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``)."""
def load_catalog(catalog_dir: Optional[Path] = None) -> List[PluginCatalogEntry]:
"""Every valid ``*.yaml`` entry in the catalog dir (``removed.yaml`` excluded), sorted by file name.
Malformed entries are skipped with a warning — never raises."""
root = catalog_dir or get_catalog_dir()
if not root.is_dir():
return []
entries: List[PluginCatalogEntry] = []
entries = []
for path in sorted(root.glob("*.yaml")):
if path.name == "removed.yaml":
continue
entry = _parse_entry(path)
data = _read_yaml(path)
entry = entry_from_mapping(data, str(path)) if data is not None else None
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 load_removed_list(catalog_dir: Optional[Path] = None) -> List[RemovedEntry]:
"""``removed.yaml``'s ``removed:`` list; missing/malformed → empty."""
path = (catalog_dir or get_catalog_dir()) / "removed.yaml"
data = _read_yaml(path) if path.is_file() else None
return _removed_from_list(data.get("removed")) if isinstance(data, dict) else []
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 get_catalog_entry(name: str, catalog_dir: Optional[Path] = None) -> Optional[PluginCatalogEntry]:
return next((e for e in load_catalog(catalog_dir) if e.name == name), None)
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.
"""
def filter_entries(entries: List[PluginCatalogEntry], query: str) -> List[PluginCatalogEntry]:
"""Case-insensitive substring match over name, description and declared tools; empty query = all."""
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
return [e for e in entries
if any(q in h.lower() for h in (e.name, e.description, *e.capabilities.provides_tools))]
# ─── Removed / blocklist ─────────────────────────────────────────────────────
def search_catalog(query: str) -> List[PluginCatalogEntry]:
return filter_entries(load_catalog(), query)
# ── 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).
def find_removed(name_or_repo: str, catalog_dir: Optional[Path] = None) -> Optional[RemovedEntry]:
"""Match a catalog name or repo URL (``.git``/trailing-slash insensitive) against the kill 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).
The in-tree list and the live-fetched list are UNIONED: a removal published after this checkout
shipped must still block, and a stale live cache must not un-block an in-tree removal.
"""
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):
for entry in load_removed_list(catalog_dir) + (live_removed_list() if catalog_dir is None else []):
if candidate == entry.name or (entry.repo and candidate_repo == _normalize_repo(entry.repo)):
return entry
return None
# ─── Live index ──────────────────────────────────────────────────────────────
# ── Live catalog ─────────────────────────────────────────────────────────────
# 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:
def _live_cache_path() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home() / "cache" / "plugin-catalog"
return get_hermes_home() / "cache" / "plugin-catalog.json"
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
def fetch_live_catalog(*, force: bool = False) -> Optional[Dict[str, Any]]:
"""The published ``plugin-catalog.json`` (``{"entries": [...], "removed": [...]}``), cached under
``HERMES_HOME/cache`` for :data:`LIVE_CATALOG_TTL_SECONDS`. ``None`` on ANY failure — callers fall
back to the in-tree catalog."""
cache = _live_cache_path()
try:
if not force and cache.is_file() and time.time() - cache.stat().st_mtime < LIVE_CATALOG_TTL_SECONDS:
return json.loads(cache.read_text(encoding="utf-8"))
except Exception as exc:
logger.debug("Plugin catalog: unreadable live cache %s: %s", cache, exc)
try:
import httpx
resp = httpx.get(
_LIVE_INDEX_URL,
timeout=_REQUEST_TIMEOUT,
follow_redirects=True,
headers={"Accept": "application/vnd.github+json"},
)
resp = httpx.get(LIVE_CATALOG_URL, timeout=_REQUEST_TIMEOUT, follow_redirects=True)
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
if len(resp.content) > _MAX_LIVE_BYTES:
raise ValueError("live catalog payload too large")
data = resp.json()
if not isinstance(data, dict) or not isinstance(data.get("entries"), list):
raise ValueError("unexpected live catalog payload")
cache.parent.mkdir(parents=True, exist_ok=True)
cache.write_text(json.dumps(data), encoding="utf-8")
return data
except Exception as exc:
logger.debug("Plugin catalog: live index fetch failed: %s", exc)
return None
logger.debug("Plugin catalog: live fetch failed: %s", exc)
try: # stale cache still beats the in-tree copy when the network is down
return json.loads(cache.read_text(encoding="utf-8")) if cache.is_file() else None
except Exception:
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)
"""Entries from the live (or cached) catalog, else the in-tree catalog."""
data = fetch_live_catalog()
if data is not None:
entries = [e for i, raw in enumerate(data["entries"])
if (e := entry_from_mapping(raw, f"{LIVE_CATALOG_URL}#{i}")) is not None]
if entries:
return entries
return load_catalog()
# ─── Human summaries ─────────────────────────────────────────────────────────
def live_removed_list() -> List[RemovedEntry]:
data = fetch_live_catalog()
return _removed_from_list(data.get("removed")) if data else []
def get_live_catalog_entry(name: str) -> Optional[PluginCatalogEntry]:
return next((e for e in load_catalog_live() if e.name == name), None)
# ── 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."""
"""One paragraph shown at install/enable prompts: what the user is 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})",
]
parts = [f"{label} {', '.join(items)}" for label, items in (
("registers tool(s):", caps.provides_tools), ("hook(s):", caps.provides_hooks),
("middleware:", caps.provides_middleware), ("requires env var(s):", caps.requires_env)) if items]
bits = [f"{entry.name} ({entry.tier}, maintained by {entry.maintainer})"]
if entry.description:
bits.append(entry.description)
bits.append(f"This plugin {capability_text}.")
bits.append(f"This plugin {'; '.join(parts) if parts else 'declares no tools, hooks, middleware, or env vars'}.")
if entry.platforms:
bits.append(f"Platforms: {', '.join(entry.platforms)}.")
if entry.requires_hermes:

View File

@@ -45,7 +45,7 @@ class PackPluginEntry:
@property
def install_identifier(self) -> Optional[str]:
"""Identifier for the install path; None for bare names (resolved via the community index)."""
"""Identifier for the install path; None for bare names (resolved via the plugin catalog)."""
if self.repo:
return f"{self.repo}/{self.subdir}" if self.subdir else self.repo
return None
@@ -199,31 +199,31 @@ class ResolvedPackPlugin:
def resolve_pack_plugins(pack: PluginPack) -> List[ResolvedPackPlugin]:
"""Resolve every entry; bare names go through the community index. Failures do not raise —
"""Resolve every entry; bare names go through the curated plugin catalog. Failures do not raise —
they are carried per-entry so the review screen shows them and install reports partial failure."""
resolved: List[ResolvedPackPlugin] = []
index_entries = None
catalog_entries = None
for entry in pack.plugins:
if entry.install_identifier is not None:
resolved.append(ResolvedPackPlugin(entry=entry, identifier=entry.install_identifier))
continue
try:
from hermes_cli.plugin_index import load_index, resolve_name
if index_entries is None:
index_entries, _src = load_index()
match, candidates = resolve_name(index_entries, entry.name or "")
except Exception as exc: # index load must not crash pack handling
if catalog_entries is None:
from hermes_cli.plugin_catalog import load_catalog_live
catalog_entries = load_catalog_live()
except Exception as exc: # catalog load must not crash pack handling
resolved.append(ResolvedPackPlugin(
entry=entry, identifier=None, resolve_error=f"community index unavailable: {exc}"))
entry=entry, identifier=None, resolve_error=f"plugin catalog unavailable: {exc}"))
continue
match = next((e for e in catalog_entries if e.name == (entry.name or "")), None)
if match is None:
detail = "ambiguous" if len(candidates) > 1 else "not found"
resolved.append(ResolvedPackPlugin(
entry=entry, identifier=None, resolve_error=f"{detail} in the community index"))
entry=entry, identifier=None, resolve_error="not found in the plugin catalog"))
continue
caps = match.capabilities
resolved.append(ResolvedPackPlugin(
entry=entry, identifier=match.install_identifier,
index_capabilities=list(match.capabilities)))
index_capabilities=[*caps.provides_tools, *caps.provides_hooks, *caps.provides_middleware]))
return resolved

View File

@@ -215,6 +215,11 @@ class RecordingContext:
def register_cli_command(self, name, *args, **kwargs):
recorded["commands"].append(str(name))
def get_config(self, key, default=None):
# Mirrors PluginContext.get_config with no config on disk: the DEFAULT, never None —
# plugins do `int(ctx.get_config("timeout", 180))` in register().
return default
def __getattr__(self, _name):
# Any other registration surface (platforms, providers, skills,
# context engines, ...) is accepted as a no-op — the probe only

View File

@@ -660,65 +660,40 @@ def _install_plugin_core(
return target, installed_manifest, installed_manifest.get("name") or target.name
def _looks_like_bare_index_name(identifier: str) -> bool:
"""True for a bare plugin name (no slash, no URL scheme) — resolved via the community index."""
return "/" not in identifier and "\\" not in identifier and not identifier.startswith(_URL_SCHEMES)
def _resolve_index_name(identifier: str, console) -> tuple[str, Optional[str]]:
"""Resolve a bare plugin name to ``(install_identifier, pinned_ref)``; exit 1 when unknown or
ambiguous. The ref is only pinned when it is an exact 40-char SHA; tags are advisory output."""
from hermes_cli.plugin_index import SECURITY_FOOTER, load_index, resolve_name
entries, source = load_index()
entry, candidates = resolve_name(entries, identifier)
if entry is None:
if len(candidates) > 1:
console.print(
f"[red]Error:[/red] Plugin name '{identifier}' is ambiguous in the "
f"community index ({source}). Candidates:")
for c in candidates:
console.print(f" {c.name} → {c.install_identifier}")
_fail(console, "Re-run with the exact name or the owner/repo identifier.")
_fail(console, (
f"[red]Error:[/red] Plugin '{identifier}' was not found in the "
f"community index ({source}). Use `hermes plugins search <term>` to "
"browse, or install directly with an owner/repo identifier."))
pinned_ref: Optional[str] = None
if entry.ref and _EXACT_COMMIT_RE.fullmatch(entry.ref):
pinned_ref = entry.ref.lower()
elif entry.ref:
console.print(
f"[dim]Index pins ref '{entry.ref}' (not an exact commit SHA); "
"installing the default branch head instead.[/dim]")
console.print(
f"[dim]Resolved '{entry.name}' via community index ({source}) → "
f"{entry.install_identifier}"
+ (f" @ {pinned_ref[:12]}[/dim]" if pinned_ref else "[/dim]"))
console.print(f"[dim]{SECURITY_FOOTER}[/dim]")
return entry.install_identifier, pinned_ref
def cmd_install(
identifier: str,
force: bool = False,
enable: Optional[bool] = None,
ref: Optional[str] = None,
allow_removed: bool = False,
) -> None:
"""Install a plugin from a Git URL, owner/repo shorthand, or index name.
"""Install a plugin from the curated catalog (bare name), a Git URL, or owner/repo shorthand.
Bare names resolve through the community index (an explicit ``--ref`` beats the index pin).
A catalog hit installs the reviewed pinned SHA (an explicit ``--ref`` wins) and records provenance in
a ``.hermes-catalog.json`` sidecar; URLs/shorthand are flagged as custom (unreviewed) sources. Every
install is checked against the catalog kill list unless *allow_removed*.
*enable* None prompts "Enable now? [y/N]"; True/False skip the prompt.
"""
from hermes_cli import plugins_cmd_catalog as catalog
console = _console()
if _looks_like_bare_index_name(identifier):
identifier, index_ref = _resolve_index_name(identifier, console)
if ref is None:
ref = index_ref
entry = None
if catalog.looks_like_catalog_name(identifier):
entry = catalog.resolve_catalog_name(identifier, console)
identifier = entry.install_identifier
console.print(f"[bold]{entry.name}[/bold] [cyan]\\[{entry.tier}][/cyan] [dim]pinned @ {entry.sha[:8]}[/dim]")
console.print(catalog.entry_capability_summary(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 catalog kill-list check. "
"This plugin may have been removed for security reasons.[/red]")
try:
git_url, _subdir = _resolve_git_url(identifier)
except ValueError as e:
if not allow_removed:
catalog.raise_if_removed(identifier, git_url, *((entry.name,) if entry else ()))
except (ValueError, PluginOperationError) as e:
_fail(console, f"[red]Error:[/red] {e}")
if git_url.startswith(("http://", "file://")):
console.print(
@@ -736,8 +711,12 @@ def cmd_install(
return _is_tty() and _ask_yes(" Install anyway? Only continue if you trust the source. [y/N]: ")
try:
target, installed_manifest, installed_name = _install_plugin_core(
identifier, force=force, ref=ref, scan_decision_cb=_interactive_scan_decision)
if entry is not None:
target, installed_manifest, installed_name = catalog.install_catalog_entry(
entry, force=force, ref=ref, allow_removed=True, scan_decision_cb=_interactive_scan_decision)
else:
target, installed_manifest, installed_name = _install_plugin_core(
identifier, force=force, ref=ref, scan_decision_cb=_interactive_scan_decision)
except PluginOperationError as e:
_fail(console, f"[red]{'Blocked' if isinstance(e, PluginScanBlocked) else 'Error'}:[/red] {e}")
if not _looks_like_plugin_dir(target):
@@ -794,8 +773,13 @@ def _pull_plugin_update(target: Path, pinned_msg, not_git_msg, before_pull=None)
def cmd_update(name: str) -> None:
"""Update an installed plugin by pulling latest from its git remote."""
from rich.markup import escape
from hermes_cli import plugins_cmd_catalog as catalog
console = _console()
target = _require_installed_plugin(name, _plugins_dir(), console)
sidecar = catalog.read_catalog_sidecar(target)
if sidecar: # catalog installs re-pin to the reviewed SHA — never `git pull`
catalog.cmd_update_catalog(name, target, sidecar, console)
return
try:
output = _pull_plugin_update(
target,
@@ -1321,18 +1305,21 @@ def cmd_list(args: Any | None = None) -> None:
enabled = _get_enabled_set()
disabled = _get_disabled_set()
entries = _filter_plugin_entries(entries, args, enabled, disabled)
from hermes_cli import plugins_cmd_catalog as catalog
# Source shows catalog provenance (``catalog:<tier>@<sha8>``); a kill-listed install is flagged.
rows = [
(name, _plugin_status(name, enabled, disabled, key=key), str(version), description, source)
(name, _plugin_status(name, enabled, disabled, key=key), str(version), description,
catalog.catalog_annotation(_dir) or source, catalog.removed_annotation(name, _dir))
for name, version, description, source, _dir, key in entries
]
if getattr(args, "json", False):
keys = ("name", "status", "version", "description", "source")
keys = ("name", "status", "version", "description", "source", "removed")
print(json.dumps([dict(zip(keys, row)) for row in rows], indent=2))
return
if getattr(args, "plain", False):
for name, status, version, _description, source in rows:
for name, status, version, _description, source, _removed in rows:
print(f"{status:12} {source:8} {version:8} {name}")
return
@@ -1343,11 +1330,17 @@ def cmd_list(args: Any | None = None) -> None:
table = _table(
(("Name", "bold"), ("Status", None), ("Version", "dim"), ("Description", None), ("Source", "dim")),
title="Plugins", show_lines=False)
for name, status_name, version, description, source in rows:
removed_lines = []
for name, status_name, version, description, source, removed in rows:
status = _STATUS_MARKUP.get(status_name, "[yellow]not enabled[/yellow]")
if removed:
name = f"[red]{name} ✗[/red]"
removed_lines.append(f"[red]✗ {name}[/red] was removed from the plugin catalog: {removed}")
table.add_row(name, status, 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")
@@ -1684,16 +1677,36 @@ def _run_composite_fallback(plugin_keys, plugin_labels, plugin_selected, disable
print()
def dashboard_install_plugin(identifier: str, *, force: bool, enable: bool) -> dict[str, Any]:
"""Non-interactive install for the web dashboard. Returns a JSON-serializable dict."""
def dashboard_install_plugin(
identifier: str, *, force: bool, enable: bool, catalog_name: Optional[str] = None,
) -> dict[str, Any]:
"""Non-interactive install for the dashboard/TUI. *catalog_name* installs a curated entry at its
pinned SHA (identifier may be empty); every path enforces the kill list (no GUI bypass)."""
from hermes_cli import plugins_cmd_catalog as catalog
warnings: list[str] = []
entry = None
if catalog_name:
entry = catalog.get_live_catalog_entry(catalog_name)
if entry is None:
return {"ok": False, "error": f"'{catalog_name}' is not in the Hermes plugin catalog."}
identifier = entry.install_identifier
else:
warnings.append("Custom (unreviewed) source — not from the Hermes catalog.")
try:
if _resolve_git_url(identifier)[0].startswith(("http://", "file://")):
git_url = _resolve_git_url(identifier)[0]
if git_url.startswith(("http://", "file://")):
warnings.append("Insecure URL scheme; prefer https:// or git@ for production installs.")
catalog.raise_if_removed(identifier, git_url, *((entry.name,) if entry else ()))
except ValueError:
pass
except PluginOperationError as exc:
return {"ok": False, "error": str(exc)}
try:
target, installed_manifest, installed_name = _install_plugin_core(identifier, force=force)
if entry is not None:
target, installed_manifest, installed_name = catalog.install_catalog_entry(
entry, force=force, allow_removed=True)
else:
target, installed_manifest, installed_name = _install_plugin_core(identifier, force=force)
except PluginScanBlocked as exc:
fields = ("pattern_id", "severity", "category", "file", "line", "description")
return {
@@ -1799,10 +1812,15 @@ def _user_installed_plugin_dir(name: str) -> Optional[Path]:
def dashboard_update_user_plugin(name: str) -> dict[str, Any]:
"""``git pull`` inside ``~/.hermes/plugins/<name>``."""
from hermes_cli import plugins_cmd_catalog as catalog
target = _user_installed_plugin_dir(name)
if target is None:
return {"ok": False, "error": f"Plugin '{name}' was not found under {_plugins_dir()}."}
sidecar = catalog.read_catalog_sidecar(target)
try:
if sidecar:
sha, changed = catalog.repin_catalog_plugin(target, sidecar)
return {"ok": True, "name": name, "sha": sha, "unchanged": not changed}
msg = _pull_plugin_update(
target,
lambda rec: (
@@ -1954,44 +1972,16 @@ def cmd_plugin_doctor(target: str = ".", *, ci: bool = False) -> None:
raise SystemExit(1)
def cmd_search(
term: str = "",
*,
json_output: bool = False,
capability: Optional[str] = None,
refresh: bool = False,
) -> None:
"""Search the community plugin index (fuzzy on name/description/tags)."""
from hermes_cli.plugin_index import SECURITY_FOOTER, load_index, search_index
console = _console()
entries, source = load_index(refresh=refresh)
results = search_index(entries, term, capability=capability)
if json_output:
print(json.dumps(
{"source": source, "query": term, "results": [e.to_dict() for e in results], "note": SECURITY_FOOTER},
indent=2))
return
if not results:
console.print(f"[yellow]No plugins matched '{term}'[/yellow] [dim](index source: {source})[/dim]")
return
table = _table(
(("Name", "bold"), ("Description", None), ("Author", None), ("Tags", "dim")),
title=f"Community plugins ({len(results)} match{'es' if len(results) != 1 else ''})")
for e in results:
desc = e.description if len(e.description) <= 70 else e.description[:67] + "..."
table.add_row(e.name, desc, e.author, ", ".join(e.tags))
console.print(table)
console.print(f"[dim]Index source: {source}. Install: hermes plugins install <name>[/dim]")
console.print(f"[dim]{SECURITY_FOOTER}[/dim]")
def _tri_state_flag(args, yes_attr: str, no_attr: str) -> Optional[bool]:
"""Map an argparse ``--x`` / ``--no-x`` pair to True / False / None (neither given)."""
return True if getattr(args, yes_attr, False) else (False if getattr(args, no_attr, False) else None)
def _catalog():
from hermes_cli import plugins_cmd_catalog
return plugins_cmd_catalog
def _action_pack(args):
from hermes_cli.plugin_packs import pack_command
pack_command(args)
@@ -2039,12 +2029,12 @@ _PLUGIN_ACTIONS = {
args.identifier,
force=getattr(args, "force", False),
enable=_tri_state_flag(args, "enable", "no_enable"),
ref=getattr(args, "ref", None)),
"search": lambda args: cmd_search(
getattr(args, "term", "") or "",
json_output=getattr(args, "json", False),
capability=getattr(args, "capability", None),
refresh=getattr(args, "refresh", False)),
ref=getattr(args, "ref", None),
allow_removed=getattr(args, "allow_removed", False)),
"search": lambda args: _catalog().cmd_search(
getattr(args, "term", "") or "", json_output=getattr(args, "json", False)),
"browse": lambda args: _catalog().cmd_search(""),
"validate": lambda args: _catalog().cmd_validate(args.path, as_json=getattr(args, "json", False)),
"update": lambda args: cmd_update(args.name),
"remove": lambda args: cmd_remove(args.name),
"rm": lambda args: cmd_remove(args.name),
@@ -2060,7 +2050,7 @@ _PLUGIN_ACTIONS = {
"compat": lambda args: cmd_compat(args),
"pack": _action_pack,
"show": lambda args: cmd_show(args.name),
"info": lambda args: cmd_show(args.name),
"info": lambda args: _catalog().cmd_info(args.name),
None: lambda args: cmd_toggle(),
}

View File

@@ -0,0 +1,299 @@
"""``hermes plugins`` catalog surface: resolution, provenance sidecar, search/browse/info/validate,
catalog-aware update, plus the dashboard/TUI-facing catalog payload helpers.
Sibling of :mod:`hermes_cli.plugins_cmd` (the installer core, enable/disable state and console helpers
live there and are imported late — this module is imported BY ``plugins_cmd``).
"""
from __future__ import annotations
import datetime
import json
import logging
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from hermes_cli.plugin_catalog import (
PluginCatalogEntry, entry_capability_summary, filter_entries, find_removed, get_live_catalog_entry,
load_catalog_live, load_removed_list, _NAME_RE,
)
logger = logging.getLogger(__name__)
CATALOG_SIDECAR = ".hermes-catalog.json"
# ── Resolution / provenance ──────────────────────────────────────────────────
def looks_like_catalog_name(identifier: str) -> bool:
"""Bare ``[a-z0-9_-]`` token — not a URL, ``owner/repo`` or path."""
from hermes_cli.plugins_cmd import _URL_SCHEMES
return bool(identifier) and "/" not in identifier and "\\" not in identifier \
and not identifier.startswith(_URL_SCHEMES) and bool(_NAME_RE.match(identifier))
def raise_if_removed(*candidates: str) -> None:
"""``PluginOperationError`` when any candidate (name or repo URL) is on the kill list."""
from hermes_cli.plugins_cmd import PluginOperationError
for candidate in candidates:
removed = find_removed(candidate)
if removed is not None:
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 catalog and is blocked from "
f"installation: {detail}")
def resolve_catalog_name(identifier: str, console) -> PluginCatalogEntry:
"""Bare name → live catalog entry, or exit 1 with a pointer to ``search``."""
from hermes_cli.plugins_cmd import _fail
entry = get_live_catalog_entry(identifier)
if entry is None:
_fail(console, (
f"[red]Error:[/red] '{identifier}' is not in the Hermes plugin catalog and is not a Git URL or "
"owner/repo shorthand. Browse entries with `hermes plugins search`."))
raise SystemExit(1) # _fail exits; keeps type-checkers honest
return entry
def write_catalog_sidecar(target: Path, entry: PluginCatalogEntry) -> None:
"""``.hermes-catalog.json`` inside the install dir — how ``update``/``list``/dashboards know the plugin
came from the catalog and at which pin."""
sidecar = {
"catalog_name": entry.name, "repo": entry.repo, "sha": entry.sha, "tier": entry.tier,
"installed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
.replace("+00:00", "Z"),
}
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) -> Optional[dict]:
"""Parsed sidecar, or ``None`` (absent/corrupt = a non-catalog install)."""
path = Path(plugin_dir) / CATALOG_SIDECAR if plugin_dir else None
if path is None or not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
return data if isinstance(data, dict) and data.get("catalog_name") else None
def catalog_annotation(dir_path) -> Optional[str]:
"""``catalog:<tier>@<sha8>`` for a catalog install (``list`` Source column), else ``None``."""
sidecar = read_catalog_sidecar(dir_path)
if not sidecar:
return None
return f"catalog:{sidecar.get('tier') or 'community'}@{str(sidecar.get('sha') or '')[:8]}"
def removed_annotation(name: str, dir_path) -> Optional[str]:
"""Kill-list reason when an INSTALLED plugin matches by name, catalog name or repo, else ``None``."""
sidecar = read_catalog_sidecar(dir_path) or {}
for candidate in (name, sidecar.get("catalog_name"), sidecar.get("repo")):
removed = find_removed(str(candidate)) if candidate else None
if removed is not None:
return removed.reason or "no reason recorded"
return None
# ── Catalog-aware install / update ───────────────────────────────────────────
def install_catalog_entry(entry: PluginCatalogEntry, *, force: bool, ref: Optional[str] = None,
allow_removed: bool = False, scan_decision_cb=None) -> tuple:
"""``_install_plugin_core`` at the catalog pin (an explicit *ref* wins) + provenance sidecar.
Returns the core's ``(target, manifest, installed_name)``."""
from hermes_cli.plugins_cmd import _install_plugin_core
if not allow_removed:
raise_if_removed(entry.name, entry.repo)
target, manifest, installed_name = _install_plugin_core(
entry.install_identifier, force=force, ref=ref or entry.sha, scan_decision_cb=scan_decision_cb)
write_catalog_sidecar(target, entry)
return target, manifest, installed_name
def repin_catalog_plugin(target: Path, sidecar: dict) -> tuple[str, bool]:
"""Re-pin a catalog install to the current catalog SHA (never ``git pull``). Returns
``(new_sha, changed)``; raises ``PluginOperationError`` when the entry left the catalog."""
from hermes_cli.plugins_cmd import PluginOperationError, _get_enabled_set, _save_enabled_set
catalog_name = str(sidecar["catalog_name"])
entry = get_live_catalog_entry(catalog_name)
if entry is None:
raise PluginOperationError(
f"Plugin '{catalog_name}' is no longer in the catalog — it may have been removed. "
"See `hermes plugins info` and the removed blocklist.")
if str(sidecar.get("sha") or "").strip().lower() == entry.sha:
return entry.sha, False
was_enabled = _get_enabled_set() # the force reinstall must not flip activation state
install_catalog_entry(entry, force=True)
_save_enabled_set(was_enabled)
return entry.sha, True
def cmd_update_catalog(name: str, target: Path, sidecar: dict, console) -> None:
from hermes_cli.plugins_cmd import PluginOperationError, _fail
console.print(f"[dim]Checking catalog pin for {name}...[/dim]")
try:
sha, changed = repin_catalog_plugin(target, sidecar)
except PluginOperationError as exc:
_fail(console, f"[red]Error:[/red] {exc}")
raise SystemExit(1)
verb = "updated to" if changed else "is already at catalog pin"
console.print(f"[green]✓[/green] Plugin [bold]{name}[/bold] {verb} {sha[:8]}.")
# ── search / browse / info / validate ────────────────────────────────────────
def _capability_counts(entry: PluginCatalogEntry) -> str:
caps = entry.capabilities
parts = [f"{len(items)} {label}{'s' if len(items) != 1 and label != 'middleware' else ''}"
for items, label in ((caps.provides_tools, "tool"), (caps.provides_hooks, "hook"),
(caps.provides_middleware, "middleware")) if items]
if caps.requires_env:
parts.append(f"{len(caps.requires_env)} env")
return ", ".join(parts) or "—"
def _render_entries(entries: List[PluginCatalogEntry], console) -> None:
from hermes_cli.plugins_cmd import _table
table = _table(((("Name", "bold")), ("Tier", None), ("Description", None), ("Pinned", "dim"),
("Capabilities", "dim")), title="Hermes Plugin Catalog (curated)")
for e in entries:
tier = "[cyan]official[/cyan]" if e.tier == "official" else "[magenta]community[/magenta]"
desc = e.description if len(e.description) <= 60 else e.description[:57] + "..."
table.add_row(e.name, tier, desc, e.sha[:8], _capability_counts(e))
console.print()
console.print(table)
console.print()
console.print("[dim]Details:[/dim] hermes plugins info <name> [dim]Install:[/dim] hermes plugins install <name>")
def cmd_search(term: str = "", *, json_output: bool = False) -> None:
"""Search the curated catalog (name/description/declared tools); empty term = browse everything."""
from hermes_cli.plugins_cmd import _console
matches = filter_entries(load_catalog_live(), term)
if json_output:
print(json.dumps({"query": term, "results": [e.to_dict() for e in matches]}, indent=2))
return
console = _console()
if not matches:
console.print(f"[yellow]No catalog entries matched '{term}'[/yellow]" if term
else "[dim]No catalog entries available.[/dim]")
return
_render_entries(matches, console)
def cmd_info(name: str) -> None:
"""Full catalog entry for *name*; falls back to installed-plugin details for non-catalog names."""
from hermes_cli.plugins_cmd import _console, cmd_show
entry = get_live_catalog_entry(name)
if entry is None:
cmd_show(name)
return
console = _console()
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()
rows = [("Repo", entry.repo), ("Subdir", entry.subdir), ("Pinned SHA", entry.sha),
("Maintainer", entry.maintainer), ("Requires", f"hermes {entry.requires_hermes}" if entry.requires_hermes else ""),
("Platforms", ", ".join(entry.platforms)), ("Docs", entry.docs_url)]
for label, value in rows:
if value:
console.print(f"[dim]{label + ':':<12}[/dim] {value}")
console.print()
for label, items in (("Tools", caps.provides_tools), ("Hooks", caps.provides_hooks),
("Middleware", caps.provides_middleware), ("Env vars", caps.requires_env)):
console.print(f"[dim]{label + ':':<12}[/dim] {', '.join(items) or '(none)'}")
console.print()
removed = find_removed(entry.name) or find_removed(entry.repo)
if removed is not None:
console.print(f"[red bold]✗ REMOVED from catalog: {removed.reason or 'no reason recorded'}"
f"{f' ({removed.date})' if removed.date else ''}[/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:
"""Catalog-admission validation of a plugin directory (the CI gate); exits 0/1."""
from hermes_cli.plugin_validate import validate_plugin_dir
from hermes_cli.plugins_cmd import _console
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 check_name, ok, detail in report.checks:
console.print(f"{'[green]✓[/green]' if ok else '[red]✗[/red]'} {check_name}"
+ (f" [dim]— {detail}[/dim]" if detail else ""))
for warning in report.warnings:
console.print(f"[yellow]⚠ {warning}[/yellow]")
console.print()
console.print("[green bold]Validation passed.[/green bold]" if report.ok else "[red bold]Validation failed.[/red bold]")
sys.exit(report.exit_code)
# ── Dashboard / TUI payloads ─────────────────────────────────────────────────
def installed_catalog_state(installed: Dict[str, Dict[str, Any]]) -> Dict[str, Any]:
"""Catalog entries merged with local state for the dashboard. *installed* maps every alias (name
and registry key) of a discovered plugin to ``{"dir", "runtime_status"}``. A catalog name rarely
equals the manifest name (``hermes-plugin-x`` vs ``x``), so installs are matched through the
sidecar's ``catalog_name`` first and by name only as a fallback."""
by_catalog_name: Dict[str, Dict[str, Any]] = {}
for local in installed.values():
sidecar = read_catalog_sidecar(local["dir"])
if sidecar:
by_catalog_name[str(sidecar["catalog_name"])] = {**local, "sidecar": sidecar}
entries = []
for entry in load_catalog_live():
local = by_catalog_name.get(entry.name) or installed.get(entry.name)
sidecar = local.get("sidecar") if local else None
installed_sha = str(sidecar["sha"]) if sidecar and sidecar.get("sha") else None
entries.append({
**entry.to_dict(), "sha_short": entry.sha[:7],
"capability_summary": entry_capability_summary(entry),
"installed": local is not None, "installed_sha": installed_sha,
"update_available": bool(installed_sha) and installed_sha != entry.sha,
"runtime_status": local["runtime_status"] if local else None,
})
return {
"entries": entries,
"removed": [{"name": r.name, "repo": r.repo, "reason": r.reason, "date": r.date} for r in load_removed_list()],
"generated_at": datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z"),
}
def catalog_row_fields(dir_path, pins: Dict[str, str]) -> Dict[str, Any]:
"""Provenance fields for one installed-plugin row (TUI/desktop ``plugins.manage list``): catalog
name/tier/installed SHA and, when *pins* has the entry, the current pin + ``update_available``."""
sidecar = read_catalog_sidecar(dir_path)
if not sidecar:
return {}
installed_sha = str(sidecar.get("sha") or "").lower()
row: Dict[str, Any] = {
"catalog_name": sidecar["catalog_name"], "catalog_tier": str(sidecar.get("tier") or "community"),
"installed_sha": installed_sha}
pin = pins.get(str(sidecar["catalog_name"]))
if pin:
row["catalog_sha"] = pin
row["update_available"] = bool(installed_sha) and installed_sha != pin
return row
def catalog_pins() -> Dict[str, str]:
"""``{catalog_name: pinned_sha}`` from the live catalog; empty on failure (best effort)."""
try:
return {e.name: e.sha for e in load_catalog_live()}
except Exception:
return {}

View File

@@ -16,17 +16,19 @@ def build_plugins_parser(subparsers, *, cmd_plugins: Callable) -> None:
plugins_subparsers = plugins_parser.add_subparsers(dest="plugins_action")
plugins_install = plugins_subparsers.add_parser(
"install", help="Install a plugin from a Git URL, owner/repo, or index name")
"install", help="Install a plugin from the curated catalog, a Git URL, or owner/repo")
plugins_install.add_argument(
"identifier",
help="Git URL, owner/repo shorthand (e.g. anpicasso/hermes-plugin-chrome-profiles), "
"or a bare plugin name resolved through the community index "
"(see `hermes plugins search`)")
help="Bare plugin catalog entry name (see `hermes plugins search`), Git URL, or owner/repo "
"shorthand (e.g. anpicasso/hermes-plugin-chrome-profiles)")
plugins_install.add_argument(
"--force", "-f", action="store_true", help="Remove existing plugin and reinstall")
plugins_install.add_argument(
"--ref", metavar="COMMIT_SHA",
help="Install exactly one immutable 40-character Git commit SHA")
plugins_install.add_argument(
"--allow-removed", action="store_true",
help="DANGEROUS: bypass the catalog removed-plugin blocklist check")
_install_enable_group = plugins_install.add_mutually_exclusive_group()
_install_enable_group.add_argument(
"--enable", action="store_true",
@@ -37,17 +39,18 @@ def build_plugins_parser(subparsers, *, cmd_plugins: Callable) -> None:
)
plugins_search = plugins_subparsers.add_parser(
"search", help="Search the community plugin index")
"search", help="Search the curated Hermes plugin catalog")
plugins_search.add_argument(
"term", nargs="?", default="",
help="Search term matched fuzzily against name, description, and tags "
"(omit to browse the full index)")
help="Query matched against entry names, descriptions and declared tools (omit to list the whole catalog)")
add_json_flag(plugins_search, "Print machine-readable JSON")
plugins_search.add_argument(
"--capability", metavar="CAP",
help="Filter by declared capability (e.g. tools, platform, commands)")
plugins_search.add_argument(
"--refresh", action="store_true", help="Bypass the local cache and re-fetch the index")
plugins_subparsers.add_parser("browse", help="List every curated plugin catalog entry")
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")
add_json_flag(plugins_validate, "Print machine-readable JSON (for CI)")
plugins_update = plugins_subparsers.add_parser(
"update", help="Pull latest changes for an installed plugin")

View File

@@ -170,12 +170,41 @@ def _plugin_action(result: dict, fallback_error: str, *, rescan: bool) -> dict:
return result
@router.get("/api/dashboard/plugins/catalog")
async def get_plugins_catalog(request: Request):
"""Curated plugin catalog merged with installed state (session protected)."""
_require_token(request)
def _run():
from hermes_cli.plugins_cmd import _discover_all_plugins, _get_disabled_set, _get_enabled_set
from hermes_cli.plugins_cmd_catalog import installed_catalog_state
from hermes_cli.web_server_dashboard import _plugin_runtime_status
enabled, disabled = _get_enabled_set(), _get_disabled_set()
installed = {}
for name, _v, _d, _s, dir_str, key in _discover_all_plugins():
aliases = {name, key} - {""}
info = {"dir": dir_str, "runtime_status": _plugin_runtime_status(aliases, enabled, disabled)}
installed.update({alias: info for alias in aliases})
return installed_catalog_state(installed)
try:
return await asyncio.to_thread(_run)
except Exception as exc:
_log.warning("plugins/catalog failed: %s", exc)
raise HTTPException(status_code=500, detail="Failed to build plugins catalog.") from exc
@router.post("/api/dashboard/agent-plugins/install")
async def post_agent_plugin_install(request: Request, body: _AgentPluginInstallBody):
_require_token(request)
from hermes_cli.plugins_cmd import dashboard_install_plugin
result = dashboard_install_plugin(body.identifier.strip(), force=body.force, enable=body.enable)
catalog_name = (body.catalog_name or "").strip()
identifier = body.identifier.strip()
if not identifier and not catalog_name:
raise HTTPException(status_code=400, detail="Provide an identifier or a catalog_name.")
result = dashboard_install_plugin(
identifier, force=body.force, enable=body.enable, catalog_name=catalog_name or None)
result = _plugin_action(result, "Install failed.", rescan=True)
# Strip internal paths from the response
result.pop("after_install_path", None)

View File

@@ -634,6 +634,11 @@ def _plugin_auth_hint(name: str, provides_tools: list) -> tuple:
return False, ""
def _plugin_runtime_status(aliases: set, enabled_set: set, disabled_set: set) -> str:
"""enabled / disabled / inactive for a plugin's name+key alias set (disabled wins)."""
return "disabled" if aliases & disabled_set else "enabled" if aliases & enabled_set else "inactive"
def _merged_plugins_hub(force_refresh: bool = False) -> Dict[str, Any]:
"""Agent discovery + dashboard manifests + provider picker metadata.
@@ -662,6 +667,7 @@ def _merged_plugins_hub(force_refresh: bool = False) -> Dict[str, Any]:
_get_enabled_set,
_read_manifest as _read_plugin_manifest_at,
)
from hermes_cli.plugins_cmd_catalog import removed_annotation
dashboard_list = _get_dashboard_plugins()
dash_by_name = {str(p["name"]): p for p in dashboard_list}
@@ -675,12 +681,7 @@ def _merged_plugins_hub(force_refresh: bool = False) -> Dict[str, Any]:
# Both the path-derived key (nested category plugins) and the bare manifest name
# count for enabled/disabled state, matching the runtime loader's back-compat lookup.
aliases = {name, key} if key else {name}
if aliases & disabled_set:
runtime_status = "disabled"
elif aliases & enabled_set:
runtime_status = "enabled"
else:
runtime_status = "inactive"
runtime_status = _plugin_runtime_status(aliases, enabled_set, disabled_set)
dir_path = Path(dir_str)
dm = dash_by_name.get(name)
@@ -708,6 +709,7 @@ def _merged_plugins_hub(force_refresh: bool = False) -> Dict[str, Any]:
"auth_required": auth_required,
"auth_command": auth_command,
"user_hidden": name in hidden_plugins,
"removed_reason": removed_annotation(name, dir_str),
})
agent_names = {r["name"] for r in rows}

View File

@@ -30,6 +30,18 @@ command. A hook with no concrete consumer is speculative infrastructure and is r
`plugin-llm-example`, `plugin-llm-async-example`) live in
[`hermes-example-plugins`](https://github.com/NousResearch/hermes-example-plugins), not here.
## Plugin catalog (`plugin-catalog/`, Sep 2026)
The ONLY discovery system for out-of-tree plugins. One YAML per entry, 40-hex SHA pin mandatory,
human-merged via PR (`plugin-catalog/README.md` = admission policy; `plugin-catalog-ci.yml` clones
each changed entry at its pin and runs `hermes plugins validate`). `removed.yaml` is the kill list —
every install path (CLI, dashboard, TUI) refuses matches; only the CLI has a loud `--allow-removed`.
Code: `hermes_cli/plugin_catalog.py` (loader, live refresh from
`/docs/api/plugin-catalog.json` published by the docs build, in-tree fallback),
`hermes_cli/plugins_cmd_catalog.py` (resolution, `.hermes-catalog.json` provenance sidecar,
search/info/validate, re-pin on `update`, dashboard/TUI payloads). Never add a second name index:
bare names resolve through the catalog or error.
## Plugin kinds and their discovery systems
| Kind | Where | Discovery | Notes |

View File

@@ -1328,7 +1328,9 @@ def _(rid, params: dict) -> dict:
# ─── Plugins ─────────────────────────────────────────────────────────────────
def _plugin_rows() -> list[dict]:
pc = _tools_mod("hermes_cli.plugins_cmd")
cat = _tools_mod("hermes_cli.plugins_cmd_catalog")
enabled, disabled = pc._get_enabled_set(), pc._get_disabled_set()
pins = cat.catalog_pins() # powers the desktop's "Update to <pin>" affordance
out = []
for name, version, desc, source, _dir, key in sorted(pc._discover_all_plugins()):
status = pc._plugin_status(name, enabled, disabled, key=key)
@@ -1339,7 +1341,8 @@ def _plugin_rows() -> list[dict]:
# key = canonical registry key (names collide across category dirs); portable = Agent Plugins v1.
out.append({
"name": name, "key": key, "version": str(version or ""), "description": desc or "",
"source": source, "status": status, "portable": pc._is_portable_plugin_dir(_dir)})
"source": source, "status": status, "portable": pc._is_portable_plugin_dir(_dir),
**cat.catalog_row_fields(_dir, pins)})
return out
@@ -1363,22 +1366,44 @@ def _plugins_toggle(rid, params):
def _plugins_install(rid, params):
# ``catalog_name`` alone installs a curated entry at its pinned SHA (resolved server-side, kill list
# enforced, no bypass) — same contract as the dashboard endpoint.
ident = (params.get("identifier") or params.get("repo") or "").strip()
if not ident:
return _err(rid, 4019, "plugins.install requires 'identifier' or 'repo'")
catalog_name = str(params.get("catalog_name") or "").strip()
if not ident and not catalog_name:
return _err(rid, 4019, "plugins.install requires 'identifier', 'repo', or 'catalog_name'")
result = _tools_mod("hermes_cli.plugins_cmd").dashboard_install_plugin(
ident, force=bool(params.get("force")), enable=params.get("enable", True))
ident, force=bool(params.get("force")), enable=params.get("enable", True), catalog_name=catalog_name or None)
return _ok(rid, result) if result.get("ok") else _err(rid, 5026, result.get("error") or "install failed")
_PLUGINS_ACTIONS = {"list": _plugins_list, "toggle": _plugins_toggle, "install": _plugins_install}
def _plugins_update(rid, params):
"""Catalog installs only: re-pin to the current catalog SHA (non-catalog installs update via the CLI)."""
name = (params.get("name") or "").strip()
if not name:
return _err(rid, 4019, "plugins.update requires a 'name'")
pc, cat = _tools_mod("hermes_cli.plugins_cmd"), _tools_mod("hermes_cli.plugins_cmd_catalog")
target = pc._plugins_dir() / name
sidecar = cat.read_catalog_sidecar(target) if target.is_dir() else None
if not sidecar:
return _err(rid, 4020, f"'{name}' is not a catalog install — update it via the CLI")
try:
sha, changed = cat.repin_catalog_plugin(target, sidecar)
except pc.PluginOperationError as e:
return _err(rid, 4021, str(e))
return _ok(rid, {"ok": True, "unchanged": not changed, "sha": sha})
_PLUGINS_ACTIONS = {"list": _plugins_list, "toggle": _plugins_toggle, "install": _plugins_install,
"update": _plugins_update}
@_scoped_rpc("plugins.manage", 5026, catch_resolve=False)
def _(rid, params: dict) -> dict:
"""TUI Plugins Hub backend (shares primitives with ``hermes plugins`` / the dashboard):
``list`` → {plugins, user_count, bundled_count}; ``toggle`` flips ``key``/``name`` per ``enable``;
``install`` git-clones ``identifier``/``repo`` (``force``, ``enable`` default True)."""
``install`` git-clones ``identifier``/``repo`` or a curated ``catalog_name`` (``force``, ``enable``
default True); ``update`` re-pins a catalog install to the current catalog SHA."""
return _run_action(rid, params, _PLUGINS_ACTIONS, "plugins")

View File

@@ -91,9 +91,18 @@ installs as `catalog:<tier>@<sha>` so you can see provenance at a glance.
### Names not in the catalog
A bare name that isn't a catalog entry falls back to the
[community plugin index](plugins.md) with a warning — those entries are
indexed, not reviewed. Catalog names always win when both exist.
A bare name that isn't a catalog entry is an error: there is no second,
unreviewed name index. Install such plugins by `owner/repo` or Git URL instead
(custom source, see below), or submit them to the catalog.
### Live refresh
The docs build publishes the catalog as one JSON document
(`https://hermes-agent.nousresearch.com/docs/api/plugin-catalog.json`).
`search`/`install`/`update` fetch it at most every six hours and cache it under
`~/.hermes/cache/`, so new entries and removals reach installed clients without
updating Hermes. Offline, the copy shipped with your checkout is used. Removals
from the in-tree list and the live list are always both enforced.
### Custom git URLs are different

View File

@@ -341,7 +341,7 @@ Declarative plugins are symlinked with a `nix-managed-` prefix — they coexist
hermes plugins # unified interactive UI
hermes plugins list # table: enabled / disabled / not enabled
hermes plugins search <term> # search the Hermes plugin catalog
hermes plugins install <name> # install by index name (resolved to repo @ pinned ref)
hermes plugins install <name> # install a catalog entry (repo @ reviewed pinned SHA)
hermes plugins install user/repo # install from Git, then prompt Enable? [y/N]
hermes plugins install user/repo --enable # install AND enable (no prompt)
hermes plugins install user/repo --no-enable # install but leave disabled (no prompt)
@@ -542,8 +542,8 @@ description: STT + streaming TTS + approval relay
author: hyper
version: 1.0.0
plugins:
- name: hermes-media-studio # bare community-index name…
ref: e8d59971d2b7901405b39dac7b03bdd616272d0d
- name: hermes-telegram-business # bare plugin-catalog name…
ref: e905f3bc5eeaa5a9dab9bc5155601b3ebec75757
- repo: owner/approval-relay # …or explicit owner/repo (or git URL)
ref: 8f3c2d1a9b4e5f6071829304a5b6c7d8e9f00112
subdir: plugins/relay # optional monorepo path

View File

@@ -15,8 +15,12 @@ meta counts and exit 0 so the docs build stays green. The page renders a
Outputs (both under website/static/api/, CDN-served at /docs/api/):
- ``plugins.json`` — list of catalog entries for the page
- ``plugins-meta.json`` — counts by tier + generatedAt + removedCount
- ``plugins.json`` — list of catalog entries for the page (camelCase)
- ``plugins-meta.json`` — counts by tier + generatedAt + removedCount
- ``plugin-catalog.json`` — ``{"entries": [raw YAML mappings], "removed": [...]}`` in the loader's own
schema; installed Hermes clients fetch this for live catalog refresh
(``hermes_cli.plugin_catalog.LIVE_CATALOG_URL``) so new entries and removals reach them without
updating. Emitting it here means the docs deploy IS the publish step — no second pipeline.
"""
from __future__ import annotations
@@ -123,20 +127,39 @@ def load_catalog_entries(catalog_dir: Path) -> list[dict]:
return entries
def count_removed(catalog_dir: Path) -> int:
"""Number of entries in plugin-catalog/removed.yaml (``removed:`` list)."""
def load_removed(catalog_dir: Path) -> list[dict]:
"""``removed:`` list from plugin-catalog/removed.yaml (mappings only)."""
removed_path = catalog_dir / "removed.yaml"
if not removed_path.is_file():
return 0
return []
try:
raw = yaml.safe_load(removed_path.read_text(encoding="utf-8"))
except (yaml.YAMLError, OSError) as e:
_log(f"could not read removed.yaml: {e}")
return 0
if not isinstance(raw, dict):
return 0
removed = raw.get("removed")
return len(removed) if isinstance(removed, list) else 0
return []
removed = raw.get("removed") if isinstance(raw, dict) else None
return [r for r in removed if isinstance(r, dict)] if isinstance(removed, list) else []
def count_removed(catalog_dir: Path) -> int:
return len(load_removed(catalog_dir))
def load_raw_entries(catalog_dir: Path) -> list[dict]:
"""Raw entry mappings (loader schema, snake_case) for ``plugin-catalog.json``; the client re-validates."""
entries: list[dict] = []
if not catalog_dir.is_dir():
return entries
for path in sorted(catalog_dir.glob("*.yaml")):
if path.name == "removed.yaml":
continue
try:
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
except (yaml.YAMLError, OSError):
continue
if isinstance(raw, dict) and raw.get("name") and raw.get("repo") and raw.get("sha"):
entries.append(raw)
return entries
def main(catalog_dir: Path = DEFAULT_CATALOG_DIR, output_dir: Path = DEFAULT_OUTPUT_DIR) -> int:
@@ -162,6 +185,9 @@ def main(catalog_dir: Path = DEFAULT_CATALOG_DIR, output_dir: Path = DEFAULT_OUT
json.dump(entries, f, separators=(",", ":"), ensure_ascii=False)
with open(output_dir / "plugins-meta.json", "w", encoding="utf-8") as f:
json.dump(meta, f, separators=(",", ":"), ensure_ascii=False)
with open(output_dir / "plugin-catalog.json", "w", encoding="utf-8") as f:
json.dump({"generated_at": meta["generatedAt"], "entries": load_raw_entries(catalog_dir),
"removed": load_removed(catalog_dir)}, f, separators=(",", ":"), ensure_ascii=False)
print(
f"Extracted {len(entries)} plugin catalog entries "