diff --git a/contributors/emails/jason@webdevtoday.com b/contributors/emails/jason@webdevtoday.com new file mode 100644 index 0000000000..5742aab405 --- /dev/null +++ b/contributors/emails/jason@webdevtoday.com @@ -0,0 +1,2 @@ +webdevtodayjason +# PR #58538 salvage for #64230 diff --git a/hermes_cli/plugin_dev.py b/hermes_cli/plugin_dev.py new file mode 100644 index 0000000000..6484caa0b9 --- /dev/null +++ b/hermes_cli/plugin_dev.py @@ -0,0 +1,290 @@ +"""Runtime-backed validation behind ``hermes plugins doctor``. + +The Doctor originated in #46456 / contributor PR #46457 by 峯岸 亮 +(@zapabob). This core command keeps that contribution's manifest/import/ +registration validation intent while routing every check through the current +runtime contracts instead of maintaining a parallel scanner. +""" + +from __future__ import annotations + +import inspect +import os +import shutil +import socket +import sys +import tempfile +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Literal +from unittest.mock import patch + +from hermes_constants import get_hermes_home + + +class _DoctorLoadError(RuntimeError): + """Raised when the real plugin runtime cannot load the target.""" + + +def _deny_network(*_args: Any, **_kwargs: Any) -> None: + raise RuntimeError("network access is disabled while Plugin Doctor runs") + + +@contextmanager +def _doctor_runtime(plugin_path: Path): + """Load one plugin through the real runtime and restore global state. + + This is deliberately private Doctor machinery, not a standalone plugin + test framework. Registration code executes under a temporary HERMES_HOME + with outbound socket connects blocked. + """ + temporary_home = tempfile.TemporaryDirectory(prefix="hermes-plugin-doctor-") + stack = ExitStack() + home = Path(temporary_home.name) + bundled = home / "bundled-plugins" + plugins_root = home / "plugins" + bundled.mkdir(parents=True) + plugins_root.mkdir(parents=True) + copied = plugins_root / plugin_path.name + shutil.copytree( + plugin_path, + copied, + ignore=shutil.ignore_patterns(".git", "__pycache__", ".pytest_cache", "*.pyc"), + ) + + stack.enter_context( + patch.dict( + os.environ, + { + "HERMES_HOME": str(home), + "HERMES_BUNDLED_PLUGINS": str(bundled), + "HERMES_ENABLE_PROJECT_PLUGINS": "0", + }, + clear=False, + ) + ) + stack.enter_context(patch.object(socket, "create_connection", _deny_network)) + stack.enter_context(patch.object(socket.socket, "connect", _deny_network)) + stack.enter_context(patch.object(socket.socket, "connect_ex", _deny_network)) + + from hermes_cli.plugins import PluginManager + from tools.registry import registry + + entries_before = {entry.name: entry for entry in registry._snapshot_entries()} + policy_before = dict(registry._plugin_override_policy) + modules_before = { + name + for name in sys.modules + if name == "hermes_plugins" or name.startswith("hermes_plugins.") + } + manager = PluginManager() + try: + manifests = manager._scan_directory(plugins_root, source="user") + if not manifests: + raise _DoctorLoadError( + f"Hermes discovery found no valid plugin manifest under {copied}" + ) + if len(manifests) != 1: + raise _DoctorLoadError( + f"Expected one plugin manifest, discovered {len(manifests)} under {copied}" + ) + manifest = manifests[0] + manager._load_plugin(manifest) + loaded = manager._plugins.get(manifest.key or manifest.name) + if loaded is None: + raise _DoctorLoadError("Plugin registration produced no runtime record") + if loaded.error: + raise _DoctorLoadError(f"Plugin registration failed: {loaded.error}") + if not loaded.enabled: + raise _DoctorLoadError("Plugin registration did not enable the runtime record") + yield SimpleNamespace( + manifest=manifest, + manager=manager, + registered_tools=tuple(sorted(loaded.tools_registered)), + registered_hooks=tuple(loaded.hooks_registered), + ) + finally: + entries_after = {entry.name: entry for entry in registry._snapshot_entries()} + changed_names = { + name + for name in set(entries_before) | set(entries_after) + if entries_after.get(name) is not entries_before.get(name) + } + with registry._lock: + for name in changed_names: + previous = entries_before.get(name) + if previous is None: + registry._tools.pop(name, None) + else: + registry._tools[name] = previous + registry._plugin_override_policy.clear() + registry._plugin_override_policy.update(policy_before) + if changed_names: + registry._generation += 1 + for name in list(sys.modules): + if ( + name not in modules_before + and (name == "hermes_plugins" or name.startswith("hermes_plugins.")) + ): + sys.modules.pop(name, None) + stack.close() + temporary_home.cleanup() + + +@dataclass(frozen=True) +class DoctorFinding: + level: Literal["error", "warning"] + message: str + + +@dataclass +class DoctorReport: + path: Path + manifest: Any | None = None + findings: list[DoctorFinding] = field(default_factory=list) + registered_tools: tuple[str, ...] = () + registered_hooks: tuple[str, ...] = () + + @property + def ok(self) -> bool: + return not any(finding.level == "error" for finding in self.findings) + + def error(self, message: str) -> None: + self.findings.append(DoctorFinding("error", message)) + + def warning(self, message: str) -> None: + self.findings.append(DoctorFinding("warning", message)) + + def format_text(self) -> str: + lines = [f"Plugin Doctor: {self.path}"] + if self.manifest is not None: + lines.append( + f" manifest: {self.manifest.name} " + f"{self.manifest.version or '(no version)'} ({self.manifest.kind})" + ) + for finding in self.findings: + marker = "ERROR" if finding.level == "error" else "WARN" + lines.append(f" {marker}: {finding.message}") + if self.ok: + lines.append( + " OK: runtime discovery, manifest parsing, import, and registration passed" + ) + lines.append( + f" registrations: {len(self.registered_tools)} tool(s), " + f"{len(self.registered_hooks)} hook(s)" + ) + return "\n".join(lines) + + +def resolve_plugin_path(target: str | os.PathLike[str] | None = None) -> Path: + """Resolve an explicit path or an installed/bundled plugin id.""" + raw = os.fspath(target or ".") + direct = Path(raw).expanduser() + if direct.is_dir(): + return direct.resolve() + + candidates: list[Path] = [] + user_root = get_hermes_home() / "plugins" + candidates.append(user_root / raw) + try: + from hermes_cli.plugins import get_bundled_plugins_dir + + bundled = get_bundled_plugins_dir() + candidates.extend( + [ + bundled / raw, + bundled / "platforms" / raw, + bundled / "model-providers" / raw, + ] + ) + except Exception: + pass + candidates.append(Path.cwd() / ".hermes" / "plugins" / raw) + for candidate in candidates: + if candidate.is_dir(): + return candidate.resolve() + raise FileNotFoundError( + f"Plugin {raw!r} was not found as a path or installed plugin id" + ) + + +def _accepts_var_kwargs(callback: Any) -> bool: + try: + parameters = inspect.signature(callback).parameters.values() + except (TypeError, ValueError): + return False + return any(parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters) + + +def doctor_plugin(target: str | os.PathLike[str] | None = None) -> DoctorReport: + """Validate one plugin through Hermes' real scanner and registration path.""" + try: + path = resolve_plugin_path(target) + except FileNotFoundError as exc: + report = DoctorReport(Path(os.fspath(target or ".")).expanduser()) + report.error(str(exc)) + return report + + report = DoctorReport(path) + try: + with _doctor_runtime(path) as host: + report.manifest = host.manifest + report.registered_tools = host.registered_tools + report.registered_hooks = host.registered_hooks + + from hermes_cli.plugins import VALID_HOOKS + + declared_hooks = host.manifest.provides_hooks + declared_tools = host.manifest.provides_tools + if not isinstance(declared_hooks, list): + report.error("provides_hooks must be a list") + declared_hooks = [] + if not isinstance(declared_tools, list): + report.error("provides_tools must be a list") + declared_tools = [] + + for name in declared_hooks: + if not isinstance(name, str): + report.error("provides_hooks entries must be strings") + elif name not in VALID_HOOKS: + report.error(f"unknown hook {name!r} in provides_hooks") + + for hook_name, callbacks in host.manager._hooks.items(): + if hook_name not in VALID_HOOKS: + report.error(f"registered unknown hook {hook_name!r}") + for callback in callbacks: + if not _accepts_var_kwargs(callback): + callback_name = getattr(callback, "__name__", repr(callback)) + report.error( + f"hook callback {callback_name!r} for {hook_name!r} " + "must accept **kwargs for forward compatibility" + ) + + declared_hook_names = {name for name in declared_hooks if isinstance(name, str)} + registered_hook_names = set(host.registered_hooks) + for name in sorted(declared_hook_names - registered_hook_names): + report.warning(f"manifest declares hook {name!r} but registration did not add it") + for name in sorted(registered_hook_names - declared_hook_names): + report.warning(f"registration adds hook {name!r} not listed in provides_hooks") + + declared_tool_names = {name for name in declared_tools if isinstance(name, str)} + registered_tool_names = set(host.registered_tools) + for name in sorted(declared_tool_names - registered_tool_names): + report.warning(f"manifest declares tool {name!r} but registration did not add it") + for name in sorted(registered_tool_names - declared_tool_names): + report.warning(f"registration adds tool {name!r} not listed in provides_tools") + except _DoctorLoadError as exc: + report.error(str(exc)) + except Exception as exc: + report.error(f"unexpected validation failure: {type(exc).__name__}: {exc}") + return report + + +__all__ = [ + "DoctorFinding", + "DoctorReport", + "doctor_plugin", + "resolve_plugin_path", +] diff --git a/hermes_cli/plugins_cmd.py b/hermes_cli/plugins_cmd.py index 968b775191..e0bd524db9 100644 --- a/hermes_cli/plugins_cmd.py +++ b/hermes_cli/plugins_cmd.py @@ -2128,6 +2128,18 @@ def dashboard_remove_user_plugin(name: str) -> dict[str, Any]: return {"ok": True, "name": name} +def cmd_plugin_doctor(target: str = ".", *, ci: bool = False) -> None: + """Validate one plugin through runtime discovery and registration.""" + from rich.console import Console + + from hermes_cli.plugin_dev import doctor_plugin + + report = doctor_plugin(target) + Console().print(report.format_text()) + if ci and not report.ok: + raise SystemExit(1) + + def plugins_command(args) -> None: """Dispatch hermes plugins subcommands.""" action = getattr(args, "plugins_action", None) @@ -2161,6 +2173,8 @@ def plugins_command(args) -> None: cmd_disable(args.name) elif action in {"list", "ls"}: cmd_list(args) + elif action == "doctor": + cmd_plugin_doctor(args.target, ci=getattr(args, "ci", False)) elif action is None: cmd_toggle() else: diff --git a/hermes_cli/subcommands/plugins.py b/hermes_cli/subcommands/plugins.py index 57796cf18f..9ed228599f 100644 --- a/hermes_cli/subcommands/plugins.py +++ b/hermes_cli/subcommands/plugins.py @@ -13,10 +13,10 @@ def build_plugins_parser(subparsers, *, cmd_plugins: Callable) -> None: """Attach the ``plugins`` subcommand to ``subparsers``.""" plugins_parser = subparsers.add_parser( "plugins", - help="Manage plugins — install, update, remove, list", + help="Manage and validate plugins", description=( - "Install, update, remove, or list native Hermes plugins and " - "portable Agent Plugins v1 packages. Portable packages install disabled." + "Install, update, remove, list, or validate native Hermes plugins " + "and portable Agent Plugins v1 packages. Portable packages install disabled." ), ) plugins_subparsers = plugins_parser.add_subparsers(dest="plugins_action") @@ -106,4 +106,20 @@ def build_plugins_parser(subparsers, *, cmd_plugins: Callable) -> None: "disable", help="Disable a plugin without removing it" ) plugins_disable.add_argument("name", help="Plugin name to disable") + + plugins_doctor = plugins_subparsers.add_parser( + "doctor", help="Validate a plugin with the real runtime contracts" + ) + plugins_doctor.add_argument( + "target", + nargs="?", + default=".", + help="Plugin path or installed plugin id (default: current directory)", + ) + plugins_doctor.add_argument( + "--ci", + action="store_true", + help="Exit non-zero when validation reports an error", + ) + plugins_parser.set_defaults(func=cmd_plugins) diff --git a/plugins/plugin_doctor/__init__.py b/plugins/plugin_doctor/__init__.py deleted file mode 100644 index 976c1fb09e..0000000000 --- a/plugins/plugin_doctor/__init__.py +++ /dev/null @@ -1,40 +0,0 @@ -"""Hermes plugin manifest/import validator.""" - -from __future__ import annotations - -from . import core -from .cli import plugin_doctor_command, register_cli - - -def _json_handler(fn): - def handler(values=None, **kwargs): - payload = values if isinstance(values, dict) else {} - payload.update(kwargs) - return core.to_json(fn(payload)) - - return handler - - -def register(ctx) -> None: - """Register Plugin Doctor tools, slash command, and CLI command.""" - ctx.register_tool( - name="plugin_doctor_scan", - toolset="plugin-doctor", - schema=core.SCAN_SCHEMA, - handler=_json_handler(core.scan_plugins), - check_fn=lambda: True, - description=core.SCAN_SCHEMA["description"], - ) - ctx.register_command( - "plugin-doctor", - handler=lambda raw_args: core.handle_slash(raw_args), - description="Validate Hermes plugin manifests and import/register entry points.", - args_hint="[plugins_dir]", - ) - ctx.register_cli_command( - name="plugin-doctor", - help="Validate Hermes plugin manifests and imports", - setup_fn=register_cli, - handler_fn=plugin_doctor_command, - description="Scan plugin.yaml metadata, imports, register(ctx), and duplicate names.", - ) diff --git a/plugins/plugin_doctor/cli.py b/plugins/plugin_doctor/cli.py deleted file mode 100644 index 84db188023..0000000000 --- a/plugins/plugin_doctor/cli.py +++ /dev/null @@ -1,22 +0,0 @@ -from __future__ import annotations - -from typing import Any - -from . import core - - -def register_cli(subparser) -> None: - subparser.add_argument("--plugins-dir", default=core.DEFAULT_PLUGINS_DIR) - subparser.add_argument("--no-import-check", action="store_true") - subparser.set_defaults(func=plugin_doctor_command) - - -def plugin_doctor_command(args: Any) -> int: - payload = core.scan_plugins( - { - "plugins_dir": args.plugins_dir, - "include_import_check": not args.no_import_check, - } - ) - print(core.to_json(payload)) - return 0 if payload.get("ok") else 1 diff --git a/plugins/plugin_doctor/core.py b/plugins/plugin_doctor/core.py deleted file mode 100644 index 1f0841c839..0000000000 --- a/plugins/plugin_doctor/core.py +++ /dev/null @@ -1,182 +0,0 @@ -from __future__ import annotations - -import importlib.util -import json -import sys -from pathlib import Path -from typing import Any - -REQUIRED_PLUGIN_KEYS = {"name", "version", "description", "kind"} -DEFAULT_PLUGINS_DIR = "plugins" - -SCAN_SCHEMA = { - "description": "Validate Hermes plugin manifests and import/register entry points.", - "type": "object", - "properties": { - "plugins_dir": { - "type": "string", - "description": "Directory containing plugin subdirectories.", - "default": DEFAULT_PLUGINS_DIR, - }, - "include_import_check": { - "type": "boolean", - "description": "Also import each plugin __init__.py and check for register(ctx).", - "default": True, - }, - }, -} - - -class _MetadataError(ValueError): - pass - - -def to_json(payload: dict[str, Any]) -> str: - return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) - - -def _load_yaml(path: Path) -> dict[str, Any]: - try: - import yaml # type: ignore - except Exception as exc: # pragma: no cover - dependency failure path - raise _MetadataError(f"PyYAML is required to parse {path.name}: {exc}") from exc - try: - data = yaml.safe_load(path.read_text(encoding="utf-8")) - except Exception as exc: - raise _MetadataError(f"failed to parse {path.name}: {exc}") from exc - if not isinstance(data, dict): - raise _MetadataError(f"{path.name} must contain a YAML object") - return data - - -def _plugin_dirs(root: Path) -> list[Path]: - if not root.exists(): - return [] - ignored = {"__pycache__", "hermes_test"} - return sorted( - path - for path in root.iterdir() - if path.is_dir() and not path.name.startswith(".") and path.name not in ignored - ) - - -def _validate_manifest(plugin_dir: Path) -> tuple[dict[str, Any], list[str], list[str]]: - errors: list[str] = [] - warnings: list[str] = [] - manifest_path = plugin_dir / "plugin.yaml" - if not manifest_path.is_file(): - return {}, ["missing plugin.yaml"], warnings - try: - manifest = _load_yaml(manifest_path) - except _MetadataError as exc: - return {}, [str(exc)], warnings - - missing = sorted(REQUIRED_PLUGIN_KEYS - set(manifest)) - if missing: - errors.append(f"missing required manifest key(s): {', '.join(missing)}") - if manifest.get("name") and str(manifest["name"]).replace("-", "_") != plugin_dir.name: - warnings.append( - "manifest name does not match directory name after dash/underscore normalization" - ) - for list_key in ("provides_tools", "provides_cli"): - value = manifest.get(list_key, []) - if value is not None and not isinstance(value, list): - errors.append(f"{list_key} must be a list") - return manifest, errors, warnings - - -def _check_import(plugin_dir: Path) -> tuple[bool, str]: - init_path = plugin_dir / "__init__.py" - if not init_path.is_file(): - return False, "missing __init__.py" - module_name = f"_hermes_plugin_doctor_{plugin_dir.name}" - try: - spec = importlib.util.spec_from_file_location( - module_name, - init_path, - submodule_search_locations=[str(plugin_dir)], - ) - if spec is None or spec.loader is None: - return False, "could not create import spec" - module = importlib.util.module_from_spec(spec) - previous = sys.modules.get(module_name) - sys.modules[module_name] = module - try: - spec.loader.exec_module(module) - finally: - if previous is None: - sys.modules.pop(module_name, None) - else: - sys.modules[module_name] = previous - except Exception as exc: - return False, f"import failed: {type(exc).__name__}: {exc}" - if not callable(getattr(module, "register", None)): - return False, "register(ctx) is missing or not callable" - return True, "ok" - - -def scan_plugins(values: dict[str, Any] | None = None) -> dict[str, Any]: - values = values or {} - root = Path(str(values.get("plugins_dir") or DEFAULT_PLUGINS_DIR)).expanduser() - include_import_check = bool(values.get("include_import_check", True)) - plugins: list[dict[str, Any]] = [] - duplicate_tools: dict[str, list[str]] = {} - duplicate_cli: dict[str, list[str]] = {} - - for plugin_dir in _plugin_dirs(root): - manifest, errors, warnings = _validate_manifest(plugin_dir) - tools = [str(item) for item in manifest.get("provides_tools", []) or []] - cli = [str(item) for item in manifest.get("provides_cli", []) or []] - for tool in tools: - duplicate_tools.setdefault(tool, []).append(plugin_dir.name) - for command in cli: - duplicate_cli.setdefault(command, []).append(plugin_dir.name) - - import_result: dict[str, Any] | None = None - if include_import_check: - ok, detail = _check_import(plugin_dir) - import_result = {"ok": ok, "detail": detail} - if not ok: - errors.append(detail) - - plugins.append( - { - "name": str(manifest.get("name") or plugin_dir.name), - "path": str(plugin_dir), - "manifest_ok": not any(error.startswith("missing") for error in errors), - "errors": errors, - "warnings": warnings, - "provides_tools": tools, - "provides_cli": cli, - "import": import_result, - } - ) - - tool_conflicts = {key: value for key, value in duplicate_tools.items() if len(value) > 1} - cli_conflicts = {key: value for key, value in duplicate_cli.items() if len(value) > 1} - for plugin in plugins: - for tool in plugin["provides_tools"]: - if tool in tool_conflicts: - plugin["errors"].append(f"duplicate tool name: {tool}") - for command in plugin["provides_cli"]: - if command in cli_conflicts: - plugin["errors"].append(f"duplicate CLI command: {command}") - - error_count = sum(len(plugin["errors"]) for plugin in plugins) - warning_count = sum(len(plugin["warnings"]) for plugin in plugins) - return { - "ok": error_count == 0, - "plugins_dir": str(root), - "plugin_count": len(plugins), - "error_count": error_count, - "warning_count": warning_count, - "tool_conflicts": tool_conflicts, - "cli_conflicts": cli_conflicts, - "plugins": plugins, - } - - -def handle_slash(raw_args: str) -> str: - parts = (raw_args or "").split() - plugins_dir = parts[0] if parts else DEFAULT_PLUGINS_DIR - return to_json(scan_plugins({"plugins_dir": plugins_dir})) diff --git a/plugins/plugin_doctor/plugin.yaml b/plugins/plugin_doctor/plugin.yaml deleted file mode 100644 index e5394a777b..0000000000 --- a/plugins/plugin_doctor/plugin.yaml +++ /dev/null @@ -1,13 +0,0 @@ -name: plugin-doctor -version: 0.1.0 -description: Validate Hermes plugin manifests, imports, and duplicate tool/CLI names. -author: HermesAgent -kind: standalone -platforms: - - windows - - linux - - macos -provides_tools: - - plugin_doctor_scan -provides_cli: - - plugin-doctor diff --git a/tests/hermes_cli/test_plugin_dev.py b/tests/hermes_cli/test_plugin_dev.py new file mode 100644 index 0000000000..088ec4fa53 --- /dev/null +++ b/tests/hermes_cli/test_plugin_dev.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import argparse +from pathlib import Path + +from hermes_cli.subcommands.plugins import build_plugins_parser + + +def _parse_plugins_args(*argv: str): + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command") + build_plugins_parser(subparsers, cmd_plugins=lambda args: None) + return parser.parse_args(["plugins", *argv]) + + +def test_plugins_parser_exposes_doctor() -> None: + doctor = _parse_plugins_args("doctor", "sample", "--ci") + + assert (doctor.plugins_action, doctor.target, doctor.ci) == ( + "doctor", + "sample", + True, + ) + + +def test_doctor_uses_registration_to_reject_bad_hook_and_callback_signature( + tmp_path: Path, +) -> None: + from hermes_cli.plugin_dev import doctor_plugin + + plugin = tmp_path / "bad-plugin" + plugin.mkdir() + (plugin / "plugin.yaml").write_text( + "\n".join( + [ + "name: bad-plugin", + "version: 0.1.0", + "description: broken contract", + "provides_hooks:", + " - typo_hook", + " - pre_tool_call", + ] + ) + + "\n", + encoding="utf-8", + ) + (plugin / "__init__.py").write_text( + "def callback(tool_name):\n" + " return None\n\n" + "def register(ctx):\n" + " ctx.register_hook('typo_hook', callback)\n" + " ctx.register_hook('pre_tool_call', callback)\n", + encoding="utf-8", + ) + + report = doctor_plugin(plugin) + messages = "\n".join(f.message for f in report.findings) + assert report.ok is False + assert "unknown hook 'typo_hook'" in messages + assert "must accept **kwargs" in messages + + +def test_doctor_accepts_manifest_defaults_from_runtime_parser(tmp_path: Path) -> None: + from hermes_cli.plugin_dev import doctor_plugin + + plugin = tmp_path / "minimal" + plugin.mkdir() + (plugin / "plugin.yaml").write_text("name: minimal\n", encoding="utf-8") + (plugin / "__init__.py").write_text( + "def register(ctx):\n pass\n", encoding="utf-8" + ) + + report = doctor_plugin(plugin) + assert report.ok, report.format_text() + assert report.manifest is not None + assert report.manifest.kind == "standalone" + + +def test_doctor_restores_global_tool_policy_and_module_state(tmp_path: Path) -> None: + import sys + + from hermes_cli.plugin_dev import doctor_plugin + from tools.registry import registry + + target = tmp_path / "cleanup-plugin" + target.mkdir() + (target / "plugin.yaml").write_text( + "name: cleanup-plugin\nprovides_tools: [cleanup_plugin_ping]\n", + encoding="utf-8", + ) + (target / "__init__.py").write_text( + "import json\n\n" + "def ping(args, **kwargs):\n return json.dumps({'ok': True})\n\n" + "def register(ctx):\n" + " ctx.register_tool(name='cleanup_plugin_ping', toolset='cleanup', " + "schema={'name': 'cleanup_plugin_ping', 'description': 'test', " + "'parameters': {'type': 'object'}}, handler=ping)\n", + encoding="utf-8", + ) + before_policy = dict(registry._plugin_override_policy) + before_modules = { + name + for name in sys.modules + if name == "hermes_plugins" or name.startswith("hermes_plugins.") + } + + report = doctor_plugin(target) + + assert report.ok, report.format_text() + assert report.registered_tools == ("cleanup_plugin_ping",) + assert registry.get_entry("cleanup_plugin_ping") is None + assert registry._plugin_override_policy == before_policy + after_modules = { + name + for name in sys.modules + if name == "hermes_plugins" or name.startswith("hermes_plugins.") + } + assert after_modules == before_modules + + +def test_doctor_blocks_live_network(tmp_path: Path) -> None: + from hermes_cli.plugin_dev import doctor_plugin + + plugin = tmp_path / "network-plugin" + plugin.mkdir() + (plugin / "plugin.yaml").write_text("name: network-plugin\n", encoding="utf-8") + (plugin / "__init__.py").write_text( + "import socket\n\n" + "def register(ctx):\n" + " socket.create_connection(('example.com', 443))\n", + encoding="utf-8", + ) + + report = doctor_plugin(plugin) + assert report.ok is False + assert "network access is disabled while Plugin Doctor runs" in report.format_text() diff --git a/tests/plugins/test_plugin_doctor_plugin.py b/tests/plugins/test_plugin_doctor_plugin.py deleted file mode 100644 index d1b218dbfa..0000000000 --- a/tests/plugins/test_plugin_doctor_plugin.py +++ /dev/null @@ -1,95 +0,0 @@ -from __future__ import annotations - -import json -from pathlib import Path - -from plugins.plugin_doctor import core, register - - -class _FakeContext: - def __init__(self) -> None: - self.tools = {} - self.commands = {} - self.cli_commands = {} - - def register_tool(self, name, **kwargs): - self.tools[name] = kwargs - - def register_command(self, name, **kwargs): - self.commands[name] = kwargs - - def register_cli_command(self, name, **kwargs): - self.cli_commands[name] = kwargs - - -def _write_plugin(root: Path, name: str, manifest: str, init: str = "def register(ctx):\n pass\n") -> None: - plugin = root / name - plugin.mkdir(parents=True) - (plugin / "plugin.yaml").write_text(manifest, encoding="utf-8") - (plugin / "__init__.py").write_text(init, encoding="utf-8") - - -def test_registers_tool_slash_and_cli_command() -> None: - ctx = _FakeContext() - register(ctx) - - assert "plugin_doctor_scan" in ctx.tools - assert "plugin-doctor" in ctx.commands - assert "plugin-doctor" in ctx.cli_commands - - -def test_scan_plugins_reports_valid_plugin(tmp_path: Path) -> None: - _write_plugin( - tmp_path, - "demo_plugin", - """ -name: demo-plugin -version: 0.1.0 -description: Demo plugin. -kind: standalone -provides_tools: - - demo_tool -provides_cli: - - demo -""".strip(), - ) - - payload = core.scan_plugins({"plugins_dir": str(tmp_path)}) - - assert payload["ok"] is True - assert payload["plugin_count"] == 1 - assert payload["plugins"][0]["import"]["ok"] is True - - -def test_scan_plugins_flags_missing_manifest(tmp_path: Path) -> None: - (tmp_path / "broken").mkdir() - - payload = core.scan_plugins({"plugins_dir": str(tmp_path)}) - - assert payload["ok"] is False - assert "missing plugin.yaml" in payload["plugins"][0]["errors"] - - -def test_scan_plugins_flags_duplicate_tools(tmp_path: Path) -> None: - manifest = """ -name: {name} -version: 0.1.0 -description: Demo plugin. -kind: standalone -provides_tools: - - duplicate_tool -""".strip() - _write_plugin(tmp_path, "plugin_a", manifest.format(name="plugin-a")) - _write_plugin(tmp_path, "plugin_b", manifest.format(name="plugin-b")) - - payload = core.scan_plugins({"plugins_dir": str(tmp_path), "include_import_check": False}) - - assert payload["ok"] is False - assert payload["tool_conflicts"] == {"duplicate_tool": ["plugin_a", "plugin_b"]} - - -def test_handle_slash_returns_json(tmp_path: Path) -> None: - result = json.loads(core.handle_slash(str(tmp_path))) - - assert result["ok"] is True - assert result["plugins_dir"] == str(tmp_path) diff --git a/website/docs/developer-guide/plugins/index.md b/website/docs/developer-guide/plugins/index.md index 1fdd45f679..5715b9e71d 100644 --- a/website/docs/developer-guide/plugins/index.md +++ b/website/docs/developer-guide/plugins/index.md @@ -172,11 +172,31 @@ Plus a hook that logs every tool call, and a bundled skill file. ## Step 1: Create the plugin directory +Create a directory and continue with Step 2: + ```bash mkdir -p ~/.hermes/plugins/calculator cd ~/.hermes/plugins/calculator ``` +### Validate with Plugin Doctor + +`hermes plugins doctor [path-or-id]` runs the same directory discovery, +manifest parser, namespaced import, `register(ctx)`, hook registry, and tool +registry used by Hermes itself. It reports invalid hook names, callbacks that do +not accept `**kwargs`, registration failures, and drift between declared and +registered tools/hooks. Pass `--ci` to exit non-zero on an error: + +```bash +hermes plugins doctor . --ci +``` + +Doctor uses a temporary `HERMES_HOME`, restores plugin registration state after +the check, and blocks direct Python socket connections to catch accidental +network access while registration runs. This is not a sandbox: plugin code still +executes in-process with the current user's permissions and can spawn subprocesses, +so only run Doctor on code you trust enough to import. + ## Step 2: Write the manifest Create `plugin.yaml`: diff --git a/website/docs/reference/cli-commands.md b/website/docs/reference/cli-commands.md index 4d16424524..cc8645393f 100644 --- a/website/docs/reference/cli-commands.md +++ b/website/docs/reference/cli-commands.md @@ -1380,6 +1380,7 @@ Unified plugin management — general plugins, memory providers, and context eng | `enable ` | Enable a disabled plugin. | | `disable ` | Disable a plugin without removing it. | | `list` (alias: `ls`) | List installed plugins with enabled/disabled status. | +| `doctor [path-or-id] [--ci]` | Validate a native plugin through the real manifest parser, loader, and registration path. `--ci` exits 1 on errors. | Provider plugin selections are saved to `config.yaml`: - `memory.provider` — active memory provider (empty = built-in only)