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

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]]
    diagnostics: Tuple[AgentPluginDiagnostic, ...]


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"))
    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").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)
    return AgentPluginPackage(  # skills are discovered before MCP: diagnostics keep that order
        name=manifest["name"], version=manifest.get("version", ""),
        description=manifest.get("description", ""), root=root, data_root=resolved_data,
        manifest=dict(manifest), skills=_discover_skills(root, diagnostics),
        mcp_servers=_discover_mcp(root, resolved_data, diagnostics), diagnostics=tuple(diagnostics))


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)