hermes-agent/hermes_cli/agent_plugins.py

"""Compatibility helpers for Agent Plugins v1 portable directory packages."""

from __future__ import annotations

import ipaddress
import json
import os
import re
import threading
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Mapping, Tuple
from urllib.parse import urlsplit

from agent.skill_utils import yaml_load
from hermes_platform.declaration import Declaration, parse_declaration

_HERMES_EXTENSION = "com.nousresearch.hermes"
_LIVENESS: dict[str, dict] = {}


def liveness_for(server_name: str) -> dict | None:
    """Return a copy of the portable server's liveness declaration."""
    value = _LIVENESS.get(server_name)
    return dict(value) if value is not None else None


def _set_liveness(server_name: str, value: object) -> None:
    if value is None:
        _LIVENESS.pop(server_name, None)
    elif isinstance(value, dict):
        _LIVENESS[server_name] = dict(value)
    else:
        raise AgentPluginError(f"server '{server_name}' liveness must be an object")


def _clear_liveness(server_name: str) -> None:
    _LIVENESS.pop(server_name, None)


@dataclass(frozen=True)
class AgentPluginServerDeclaration:
    declaration: Declaration
    liveness: dict | None

PLUGIN_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json"
MCP_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json"

_PLUGIN_FIELDS = {"$schema", "name", "version", "description", "author", "homepage", "repository",
                  "license", "keywords", "extensions"}
_AUTHOR_FIELDS = {"name", "email", "url"}
_STDIO_FIELDS = {"type", "command", "args", "env", "cwd"}
_REMOTE_FIELDS = {"type", "url", "headers"}
_PLUGIN_NAME_RE = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$")
_SKILL_NAME_RE = re.compile(r"^(?!.*--)[a-z0-9]+(?:-[a-z0-9]+)*$")
_PLACEHOLDER_RE = re.compile(r"\$\{(PLUGIN_ROOT|PLUGIN_DATA)\}")
_HEADER_NAME_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$")


class AgentPluginError(ValueError):
    """Fatal portable manifest validation failure."""


@dataclass(frozen=True)
class AgentPluginDiagnostic:
    scope: str
    message: str


@dataclass(frozen=True)
class AgentPluginSkill:
    name: str
    description: str
    root: Path
    skill_md: Path
    frontmatter: Mapping[str, Any]


@dataclass(frozen=True)
class AgentPluginPackage:
    name: str
    version: str
    description: str
    root: Path
    data_root: Path
    manifest: Mapping[str, Any]
    skills: tuple[AgentPluginSkill, ...]
    mcp_servers: Mapping[str, dict[str, Any]]
    server_declarations: Mapping[str, AgentPluginServerDeclaration]
    diagnostics: tuple[AgentPluginDiagnostic, ...]


_SERVER_DECLARATION_KEYS = frozenset({"app", "requires", "liveness", "trust", "timeout", "costs", "spend_limit", "cost_unit"})
# Hermes call-policy keys: alone they configure a server, they do not declare an application.
_SERVER_POLICY_KEYS = frozenset({"trust", "timeout", "costs", "spend_limit", "cost_unit"})


def _server_declarations(
    manifest: Mapping[str, Any], mcp_servers: Mapping[str, dict[str, Any]],
    diagnostics: list[AgentPluginDiagnostic] | None = None,
) -> dict[str, AgentPluginServerDeclaration]:
    namespace = manifest.get("extensions", {}).get(_HERMES_EXTENSION, {})
    raw_servers = namespace.get("servers", {})
    if not isinstance(raw_servers, dict):
        raise AgentPluginError(f"extension '{_HERMES_EXTENSION}'.servers must be an object")
    declarations: dict[str, AgentPluginServerDeclaration] = {}
    for name, raw in raw_servers.items():
        if name not in mcp_servers:
            raise AgentPluginError(f"server declaration '{name}' has no matching mcp.json server")
        if not isinstance(raw, dict):
            raise AgentPluginError(f"server declaration '{name}' must be an object")
        # A key added by a newer Hermes must not make the package uninstallable here: the older
        # release ignores it with a diagnostic, like unknown top-level fields. A key whose absence
        # would be unsafe needs the catalog entry's requires_hermes, not a hard failure on every release:
        # plugin.json has no version floor, and install/update by catalog name enforce the entry's
        # (plugins_cmd_catalog.refuse_incompatible_catalog_entry / repin_catalog_plugin).
        unknown = set(raw) - _SERVER_DECLARATION_KEYS
        if unknown:
            if diagnostics is not None:
                diagnostics.append(AgentPluginDiagnostic(
                    f"server:{name}", f"ignored unknown keys {sorted(unknown)} (needs a newer Hermes)"))
            raw = {key: value for key, value in raw.items() if key not in unknown}
            if not raw:
                continue
        if "liveness" in raw and "app" not in raw and "requires" not in raw:
            raise AgentPluginError(f"server declaration '{name}' has liveness without app or requires")
        if raw and set(raw) <= _SERVER_POLICY_KEYS:
            continue  # transport/policy settings alone do not declare an application
        try:
            declaration = parse_declaration(
                name, raw.get("app"), raw.get("requires"),
                where=f"plugin.json extension server {name!r}",
            )
        except ValueError as exc:
            raise AgentPluginError(str(exc)) from exc
        liveness = raw.get("liveness")
        if liveness is not None and not isinstance(liveness, dict):
            raise AgentPluginError(f"server declaration '{name}' liveness must be an object")
        declarations[name] = AgentPluginServerDeclaration(
            declaration=declaration,
            liveness=dict(liveness) if liveness is not None else None,
        )
    return declarations


def _server_trust(manifest: Mapping[str, Any], mcp_servers: Mapping[str, dict[str, Any]]) -> dict[str, str]:
    """Servers the package asks Hermes to gate: ``extensions.com.nousresearch.hermes.servers.<name>.trust``.

    The package may only narrow access. ``untrusted`` makes every write-capable call to that server
    ask the user first (the native ``trust: untrusted`` gate); ``full`` is the default and so changes
    nothing. A package can never widen a server the user configured: a ``config.yaml`` server with
    the same name replaces the package's entry, trust included.
    """
    raw_servers = manifest.get("extensions", {}).get(_HERMES_EXTENSION, {}).get("servers", {})
    trust: dict[str, str] = {}
    for name, raw in (raw_servers.items() if isinstance(raw_servers, dict) else ()):
        if not isinstance(raw, dict) or "trust" not in raw or name not in mcp_servers:
            continue  # shape errors are raised by _server_declarations
        value = raw["trust"]
        if value not in ("full", "untrusted"):
            raise AgentPluginError(f"server declaration '{name}' trust must be 'untrusted' or 'full', not {value!r}")
        if value == "untrusted":
            trust[name] = value
    return trust


def _server_timeouts(manifest: Mapping[str, Any], mcp_servers: Mapping[str, dict[str, Any]]) -> dict[str, float]:
    """Translate optional Hermes call deadlines without changing other servers' defaults."""
    raw_servers = manifest.get("extensions", {}).get(_HERMES_EXTENSION, {}).get("servers", {})
    timeouts: dict[str, float] = {}
    for name, raw in (raw_servers.items() if isinstance(raw_servers, dict) else ()):
        if not isinstance(raw, dict) or "timeout" not in raw or name not in mcp_servers:
            continue
        value = raw["timeout"]
        if isinstance(value, bool) or not isinstance(value, (int, float)) or not 0 < value <= threading.TIMEOUT_MAX:
            raise AgentPluginError(f"server declaration '{name}' timeout must be positive seconds within the platform wait limit")
        timeouts[name] = float(value)
    return timeouts


def _server_costs(manifest: Mapping[str, Any], mcp_servers: Mapping[str, dict[str, Any]]) -> dict[str, dict[str, Any]]:
    """Translate ``costs``/``spend_limit``/``cost_unit`` into the native ``tool_costs``/``spend_limit``/``cost_unit``
    keys (``tools.mcp_tool_costs``). A malformed table disables the package: a budget the author declared must
    never be dropped while the server still spends."""
    from tools.mcp_tool_costs import parse_cost_policy
    raw_servers = manifest.get("extensions", {}).get(_HERMES_EXTENSION, {}).get("servers", {})
    translated: dict[str, dict[str, Any]] = {}
    for name, raw in (raw_servers.items() if isinstance(raw_servers, dict) else ()):
        if not isinstance(raw, dict) or name not in mcp_servers:
            continue
        declared = {key: raw[key] for key in ("costs", "spend_limit", "cost_unit") if key in raw}
        if not declared:
            continue
        try:
            parse_cost_policy(declared.get("costs"), declared.get("spend_limit"), declared.get("cost_unit"),
                              where=f"server declaration '{name}'")
        except ValueError as exc:
            raise AgentPluginError(str(exc)) from exc
        if "costs" in declared:
            declared["tool_costs"] = declared.pop("costs")
        translated[name] = declared
    return translated


def _inside(path: Path, root: Path) -> bool:
    try:
        return path.resolve(strict=False).is_relative_to(root.resolve(strict=True))
    except (OSError, RuntimeError, ValueError):
        return False


def _all_str(values) -> bool:
    return all(isinstance(v, str) for v in values)


def _str_map(value: object) -> bool:
    return isinstance(value, dict) and _all_str(value) and _all_str(value.values())


def _read_json_object(path: Path, *, label: str) -> dict:
    try:
        value = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        raise AgentPluginError(f"{label} is not valid readable JSON: {exc}") from exc
    if not isinstance(value, dict):
        raise AgentPluginError(f"{label} must contain a JSON object")
    return value


def _validate_manifest(root: Path) -> tuple[dict, list[AgentPluginDiagnostic]]:
    manifest_path = root / "plugin.json"
    if not _inside(manifest_path, root) or not manifest_path.is_file():
        raise AgentPluginError("plugin.json must be a regular file within the plugin root")
    manifest = _read_json_object(manifest_path, label="plugin.json")
    diagnostics: list[AgentPluginDiagnostic] = []
    for field in sorted(set(manifest) - _PLUGIN_FIELDS):
        diagnostics.append(AgentPluginDiagnostic("manifest", f"ignored unknown top-level field: {field}"))
        manifest.pop(field)
    if manifest.get("$schema") != PLUGIN_SCHEMA_V1:
        raise AgentPluginError("plugin.json declares an unsupported or missing Agent Plugins schema")
    name = manifest.get("name")
    if not isinstance(name, str) or not 1 <= len(name) <= 64 or _PLUGIN_NAME_RE.fullmatch(name) is None:
        raise AgentPluginError("plugin.json name does not satisfy v1 constraints")
    for field in ("version", "description", "homepage", "repository", "license"):
        if field in manifest and not isinstance(manifest[field], str):
            raise AgentPluginError(f"plugin.json {field} must be a string")
    keywords = manifest.get("keywords", [])
    if not isinstance(keywords, list) or not _all_str(keywords):
        raise AgentPluginError("plugin.json keywords must be an array of strings")
    author = manifest.get("author", {})
    if not isinstance(author, dict):
        raise AgentPluginError("plugin.json author must be an object")
    if set(author) - _AUTHOR_FIELDS or not _all_str(author.values()):
        raise AgentPluginError("plugin.json author may contain only string name, email, and url fields")
    extensions = manifest.get("extensions", {})
    if not isinstance(extensions, dict):
        diagnostics.append(AgentPluginDiagnostic("manifest", "ignored non-object extensions field"))
        manifest.pop("extensions")
    elif any(not isinstance(value, dict) for value in extensions.values()):
        raise AgentPluginError("plugin.json extension namespace values must be objects")
    return manifest, diagnostics


def _valid_skill_frontmatter(frontmatter: Mapping[str, Any], directory_name: str) -> str | None:
    name = frontmatter.get("name")
    if (not isinstance(name, str) or name != directory_name or not 1 <= len(name) <= 64
            or _SKILL_NAME_RE.fullmatch(name) is None):
        return "name must match the directory and satisfy Agent Skills constraints"
    description = frontmatter.get("description")
    if not isinstance(description, str) or not 1 <= len(description) <= 1024:
        return "description must be a non-empty string of at most 1024 characters"
    if "license" in frontmatter and not isinstance(frontmatter["license"], str):
        return "license must be a string"
    if "compatibility" in frontmatter:
        compatibility = frontmatter["compatibility"]
        if not isinstance(compatibility, str) or not 1 <= len(compatibility) <= 500:
            return "compatibility must be a string of 1 to 500 characters"
    if "metadata" in frontmatter and not _str_map(frontmatter["metadata"]):
        return "metadata must map string keys to string values"
    if "allowed-tools" in frontmatter and not isinstance(frontmatter["allowed-tools"], str):
        return "allowed-tools must be a string"
    return None


def _parse_skill_frontmatter(skill_md: Path) -> dict:
    """Read SKILL.md and return its YAML frontmatter object; raises ValueError/OSError/UnicodeError."""
    content = skill_md.read_text(encoding="utf-8-sig").lstrip("\ufeff")
    if not content.startswith("---"):
        raise ValueError("missing YAML frontmatter")
    end_match = re.search(r"\n---\s*\n", content[3:])
    if end_match is None:
        raise ValueError("unterminated YAML frontmatter")
    try:
        parsed = yaml_load(content[3 : end_match.start() + 3])
    except Exception as exc:
        raise ValueError(f"invalid YAML frontmatter: {exc}") from exc
    if not isinstance(parsed, dict):
        raise ValueError("YAML frontmatter must be an object")
    return parsed


def _discover_skills(root: Path, diagnostics: list[AgentPluginDiagnostic]) -> tuple[AgentPluginSkill, ...]:
    skills_root = root / "skills"
    if not skills_root.exists() and not skills_root.is_symlink():
        return ()
    if not _inside(skills_root, root) or not skills_root.is_dir():
        diagnostics.append(AgentPluginDiagnostic("skills", "skills must be an in-root directory"))
        return ()
    try:
        children = sorted(skills_root.iterdir(), key=lambda path: path.name)
    except OSError as exc:
        diagnostics.append(AgentPluginDiagnostic("skills", f"cannot list skills: {exc}"))
        return ()
    skills: list[AgentPluginSkill] = []
    for child in children:
        skill_md = child / "SKILL.md"
        if not child.is_dir() or not skill_md.exists():
            continue
        if not _inside(skill_md, root) or not skill_md.is_file():
            error = "SKILL.md must be a regular in-root file"
        else:
            try:
                frontmatter = _parse_skill_frontmatter(skill_md)
            except (OSError, UnicodeError, ValueError) as exc:
                error = f"invalid SKILL.md: {exc}"
            else:
                error = _valid_skill_frontmatter(frontmatter, child.name)
        if error:
            diagnostics.append(AgentPluginDiagnostic(f"skill:{child.name}", error))
            continue
        skills.append(AgentPluginSkill(
            name=child.name, description=frontmatter["description"],
            root=child.resolve(strict=True), skill_md=skill_md.resolve(strict=True),
            frontmatter=dict(frontmatter)))
    return tuple(skills)


def _expand(value: str, plugin_root: Path, data_root: Path) -> str:
    replacements = {"PLUGIN_ROOT": str(plugin_root), "PLUGIN_DATA": str(data_root)}
    return _PLACEHOLDER_RE.sub(lambda match: replacements[match.group(1)], value)


def _resolve_scoped_path(value: str, plugin_root: Path, data_root: Path, *,
                         expand_placeholders: bool = True) -> Path:
    expanded = _expand(value, plugin_root, data_root) if expand_placeholders else value
    if value.startswith("./"):
        base, candidate = plugin_root, plugin_root / expanded[2:]
    elif value == "${PLUGIN_ROOT}" or value.startswith("${PLUGIN_ROOT}/"):
        base, candidate = plugin_root, Path(expanded)
    elif value == "${PLUGIN_DATA}" or value.startswith("${PLUGIN_DATA}/"):
        base, candidate = data_root, Path(expanded)
    else:
        raise ValueError("path must start with ./, ${PLUGIN_ROOT}, or ${PLUGIN_DATA}")
    resolved = candidate.resolve(strict=False)
    try:
        resolved.relative_to(base.resolve(strict=False))
    except (OSError, RuntimeError, ValueError) as exc:
        raise ValueError("path escapes its resolved root") from exc
    return resolved


def _validate_headers(headers: object) -> bool:
    if not isinstance(headers, dict):
        return headers is None
    names = [name for name in headers if isinstance(name, str) and _HEADER_NAME_RE.fullmatch(name)]
    values = headers.values()
    return (len(names) == len(headers) and len({name.lower() for name in names}) == len(names)
            and all(isinstance(v, str) and "\r" not in v and "\n" not in v for v in values))


def _validate_remote_url(url: object) -> str:
    """Validate a portable remote MCP URL per Agent Plugins v1 §7.2.1 and return it: absolute
    http(s), no user info, no fragment; HTTP only for ``localhost``/loopback IP. No expansion."""
    if not isinstance(url, str) or not url:
        raise ValueError("url must be a non-empty string")
    try:
        parsed = urlsplit(url)
    except ValueError as exc:
        raise ValueError(f"url is not parseable: {exc}") from exc
    scheme = parsed.scheme.lower()
    if scheme not in {"http", "https"}:
        raise ValueError("url scheme must be http or https")
    if parsed.username is not None or parsed.password is not None:
        raise ValueError("url must not contain user information")
    if parsed.fragment:
        raise ValueError("url must not contain a fragment")
    host = parsed.hostname
    if not host:
        raise ValueError("url must have a host")
    if scheme == "http" and host != "localhost":
        try:
            loopback = ipaddress.ip_address(host).is_loopback
        except ValueError:
            loopback = False
        if not loopback:
            raise ValueError("non-loopback url must use https")
    return url


def _translate_remote(config: Mapping[str, Any]) -> dict[str, Any]:
    """Translate a portable ``streamable-http`` entry into native MCP config. The v1 spec requires
    ``strict_redirect_headers``: drop configured headers on any cross-origin redirect."""
    if set(config) - _REMOTE_FIELDS:
        raise ValueError("unknown remote field")
    url = _validate_remote_url(config.get("url"))
    if not _validate_headers(config.get("headers")):
        raise ValueError("invalid headers")
    translated: dict[str, Any] = {"url": url, "strict_redirect_headers": True}
    if config.get("headers"):
        translated["headers"] = dict(config["headers"])
    return translated


def _translate_stdio(config: Mapping[str, Any], plugin_root: Path, data_root: Path,
                     create_data: bool = False) -> dict[str, Any]:
    if set(config) - _STDIO_FIELDS:
        raise ValueError("unknown stdio field")
    command = config.get("command")
    if not isinstance(command, str) or not command or "\x00" in command:
        raise ValueError("command must be a non-empty executable token")
    if command.startswith("./"):
        command = str(_resolve_scoped_path(command, plugin_root, data_root,
                                           expand_placeholders=False))
    elif any(character.isspace() for character in command):
        raise ValueError("command must contain one executable token")
    elif "/" in command or "\\" in command or command in {".", ".."}:
        raise ValueError("command must be a bare executable or begin with ./")
    args = config.get("args", [])
    if not isinstance(args, list) or not _all_str(args):
        raise ValueError("args must be an array of strings")
    env = config.get("env", {})
    if not _str_map(env):
        raise ValueError("env must map string keys to string values")
    if {"PLUGIN_ROOT", "PLUGIN_DATA"} & {key.upper() if os.name == "nt" else key for key in env}:
        raise ValueError("PLUGIN_ROOT and PLUGIN_DATA are reserved")
    cwd = config.get("cwd")
    if cwd is not None and not isinstance(cwd, str):
        raise ValueError("cwd must be a string")
    cwd_path = plugin_root if cwd is None else _resolve_scoped_path(cwd, plugin_root, data_root)
    if create_data:
        data_root.mkdir(parents=True, exist_ok=True)
        # The MCP client starts stdio servers with this cwd. Create only data-root descendants;
        # plugin-root paths remain package-owned and are never made writable by discovery.
        if cwd_path.is_relative_to(data_root):
            cwd_path.mkdir(parents=True, exist_ok=True)
    translated_env = {**{key: _expand(value, plugin_root, data_root) for key, value in env.items()},
                      "PLUGIN_ROOT": str(plugin_root), "PLUGIN_DATA": str(data_root)}
    return {"command": command, "args": [_expand(value, plugin_root, data_root) for value in args],
            "env": translated_env, "cwd": str(cwd_path)}


def _reject_sse(server: Mapping[str, Any]) -> None:
    if (set(server) - _REMOTE_FIELDS or not isinstance(server.get("url"), str)
            or not server.get("url") or not _validate_headers(server.get("headers"))):
        raise ValueError("invalid remote entry")
    raise ValueError("portable sse transport is not supported")


def _discover_mcp(root: Path, data_root: Path, diagnostics: list[AgentPluginDiagnostic], *,
                  create_data: bool = True) -> dict[str, dict[str, Any]]:
    mcp_path = root / "mcp.json"
    if not mcp_path.exists() and not mcp_path.is_symlink():
        return {}
    try:  # file-level problems (incl. AgentPluginError from the JSON read) -> one "mcp" diagnostic
        if not _inside(mcp_path, root) or not mcp_path.is_file():
            raise ValueError("mcp.json must be a regular in-root file")
        config = _read_json_object(mcp_path, label="mcp.json")
        if set(config) != {"$schema", "mcpServers"}:
            raise ValueError("mcp.json has an invalid top-level shape")
        if config.get("$schema") != MCP_SCHEMA_V1:
            raise ValueError("mcp.json declares an unsupported schema")
        servers = config.get("mcpServers")
        if not isinstance(servers, dict):
            raise ValueError("mcpServers must be an object")
    except ValueError as exc:
        diagnostics.append(AgentPluginDiagnostic("mcp", str(exc)))
        return {}
    translators = {"stdio": lambda server: _translate_stdio(server, root, data_root, create_data),
                   "streamable-http": _translate_remote, "sse": _reject_sse}
    translated: dict[str, dict[str, Any]] = {}
    for name, server in servers.items():
        try:
            if not isinstance(name, str) or not name or not isinstance(server, dict):
                raise ValueError("invalid server entry")
            translate = translators.get(server.get("type"))
            if translate is None:
                raise ValueError("unknown MCP server type")
            translated[name] = translate(server)
        except (OSError, ValueError) as exc:
            diagnostics.append(AgentPluginDiagnostic(f"mcp:{name}", str(exc)))
    return translated


def _validate_root(plugin_root: Path) -> tuple[Path, dict, list[AgentPluginDiagnostic]]:
    root = Path(plugin_root).resolve(strict=True)
    if not root.is_dir():
        raise AgentPluginError("plugin root must be a directory")
    return root, *_validate_manifest(root)


def load_agent_plugin(plugin_root: Path, data_root: Path) -> AgentPluginPackage:
    """Validate and translate one installed Agent Plugins v1 package."""
    root, manifest, diagnostics = _validate_root(plugin_root)
    resolved_data = Path(data_root).resolve(strict=False)
    skills = _discover_skills(root, diagnostics)
    mcp_servers = _discover_mcp(root, resolved_data, diagnostics)
    for name, tier in _server_trust(manifest, mcp_servers).items():
        mcp_servers[name] = {**mcp_servers[name], "trust": tier}
    for name, timeout in _server_timeouts(manifest, mcp_servers).items():
        mcp_servers[name] = {**mcp_servers[name], "timeout": timeout}
    for name, costs in _server_costs(manifest, mcp_servers).items():
        mcp_servers[name] = {**mcp_servers[name], **costs}
    server_declarations = _server_declarations(manifest, mcp_servers, diagnostics)
    return AgentPluginPackage(
        name=manifest["name"], version=manifest.get("version", ""),
        description=manifest.get("description", ""), root=root, data_root=resolved_data,
        manifest=dict(manifest), skills=skills, mcp_servers=mcp_servers,
        server_declarations=server_declarations,
        diagnostics=tuple(diagnostics),
    )


def manifest_config_schema(manifest: Any) -> Any:
    """Raw ``extensions.com.nousresearch.hermes.config_schema`` of a parsed ``plugin.json`` (Agent
    Plugins v1 forbids unknown top-level fields), else ``None``. Unvalidated: the settings form
    filters it, the loader and ``plugins validate`` report what is malformed."""
    extensions = manifest.get("extensions") if isinstance(manifest, dict) else None
    hermes = extensions.get(_HERMES_EXTENSION) if isinstance(extensions, dict) else None
    return hermes.get("config_schema") if isinstance(hermes, dict) else None


def read_agent_plugin_manifest(plugin_root: Path) -> tuple[dict, tuple[AgentPluginDiagnostic, ...]]:
    """Validate only root ``plugin.json`` without discovering components."""
    _root, manifest, diagnostics = _validate_root(plugin_root)
    return manifest, tuple(diagnostics)


def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool:
    """Import-compatible wrapper for the shared PluginManager MCP probe; directory scanning lives
    in :mod:`hermes_cli.plugins` so startup gating and full plugin discovery cannot drift apart."""
    from hermes_cli.plugins import has_enabled_agent_plugin_mcp as _probe
    return _probe(raw_config)