hermes-agent/hermes_cli/plugins_cmd.py

"""``hermes plugins`` CLI subcommand — install, update, remove, and list plugins.

After install, if the plugin ships an ``after-install.md`` file it is rendered with Rich Markdown.
Otherwise a default confirmation is shown.
"""

from __future__ import annotations
from hermes_cli.cli_output import line_input

import functools
import importlib.metadata
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import tempfile
import urllib.parse
from pathlib import Path
from typing import Any, Optional

from hermes_constants import get_hermes_home
from hermes_cli._subprocess_compat import noninteractive_git_env
from hermes_cli.config import cfg_get
from hermes_cli.secret_prompt import masked_secret_prompt
from utils import atomic_write_text

logger = logging.getLogger(__name__)


@functools.lru_cache(maxsize=1)
def _resolve_git_executable() -> Optional[str]:
    """Resolve a git binary for subprocess use when ``PATH`` may be minimal."""
    found = shutil.which("git")
    if found:
        return found
    if os.name == "nt":
        roots = [
            os.environ.get("ProgramFiles", r"C:\Program Files"),
            os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)"),
        ]
        local = os.environ.get("LOCALAPPDATA", "")
        if local:
            roots.append(os.path.join(local, "Programs"))
        candidates = [
            os.path.join(root, "Git", sub, "git.exe") for root in roots for sub in ("cmd", "bin")
        ]
    else:
        candidates = ["/usr/bin/git", "/usr/local/bin/git", "/bin/git"]
    for c in candidates:
        if c and os.path.isfile(c):
            return c
    return None


class PluginOperationError(Exception):
    """Recoverable plugin install/update failure (CLI exits; HTTP maps to 4xx)."""


class PluginScanBlocked(PluginOperationError):
    """Plugin failed the security scan and was not installed."""

    def __init__(self, message: str, scan_result=None):
        super().__init__(message)
        self.scan_result = scan_result


def _console():
    """A fresh Rich console (imported lazily; rich is not needed at import time)."""
    from rich.console import Console

    return Console()


def _is_tty() -> bool:
    return sys.stdin.isatty() and sys.stdout.isatty()


def _ask_yes(prompt: str, reader=input) -> bool:
    """One y/N question; EOF / Ctrl-C count as "no"."""
    try:
        answer = reader(prompt).strip().lower()
    except (EOFError, KeyboardInterrupt):
        return False
    return answer in {"y", "yes"}


def _sub_dict(parent: dict, key: str) -> dict:
    """``parent[key]`` as a dict, replacing a missing or non-dict value with ``{}``."""
    child = parent.get(key)
    if not isinstance(child, dict):
        child = {}
        parent[key] = child
    return child


def _config_value(*keys: str, default: Any) -> Any:
    """Read ``keys`` from config.yaml; *default* on a missing key or any load failure."""
    try:
        from hermes_cli.config import load_config
        return cfg_get(load_config(), *keys, default=default)
    except Exception:
        return default


def _config_name_set(*keys: str) -> set:
    """Read a list-valued config key as a set (empty on any failure or non-list)."""
    value = _config_value(*keys, default=[])
    return set(value) if isinstance(value, list) else set()


def _config_str(*keys: str, default: str) -> str:
    """Read a string config key, coercing empty/missing/failed reads to *default*."""
    return _config_value(*keys, default=default) or default


def _write_config_value(section: str, key: str, value: Any) -> None:
    """Persist ``config[section][key] = value`` to config.yaml (creating the section)."""
    from hermes_cli.config import load_config, save_config
    config = load_config()
    if section not in config:
        config[section] = {}
    config[section][key] = value
    save_config(config)


def _scan_on_install_enabled() -> bool:
    """Whether install/update-time plugin security scanning is enabled.

    On by default (inspired by Claude Cowork's skill & plugin security scanning). Disable via
    ``plugins.scan_on_install: false`` in config.yaml.
    """
    return bool(_config_value("plugins", "scan_on_install", default=True))


def _scan_plugin_tree(plugin_dir: Path, identifier: str, *, force: bool, scan_decision_cb=None):
    """Scan *plugin_dir* and enforce the install policy.

    Verdicts: safe → proceed; caution → needs confirmation (``force=True`` or a truthy
    ``scan_decision_cb(result)``); dangerous → always blocked. Raises :class:`PluginScanBlocked`
    when the plugin may not be installed. Returns the ScanResult (or None when scanning is
    disabled).
    """
    if not _scan_on_install_enabled():
        return None

    from tools.plugin_guard import (
        format_scan_report,
        scan_plugin,
        should_allow_plugin_install,
    )

    result = scan_plugin(plugin_dir, source=identifier)
    allowed, reason = should_allow_plugin_install(result, force=force)

    if allowed is None and scan_decision_cb is not None:
        try:
            if scan_decision_cb(result):
                allowed = True
                reason = "Caution verdict accepted by user"
        except Exception:
            logger.exception("plugin scan decision callback failed")

    if allowed is not True:
        raise PluginScanBlocked(
            f"Security scan blocked plugin install: {reason}\n\n"
            f"{format_scan_report(result)}\n"
            "Review the findings above. Install only plugins from sources "
            "you trust. (Scanning can be configured via "
            "plugins.scan_on_install in config.yaml.)",
            scan_result=result,
        )
    logger.info("plugin scan passed for %s: %s", plugin_dir.name, reason)
    return result


# Minimum manifest version this installer understands.
# Plugins may declare ``manifest_version: 1`` in plugin.yaml;
# future breaking changes to the manifest schema bump this.
_SUPPORTED_MANIFEST_VERSION = 1


def _plugins_dir() -> Path:
    """Return the user plugins directory, creating it if needed."""
    plugins = get_hermes_home() / "plugins"
    plugins.mkdir(parents=True, exist_ok=True)
    return plugins


def _sanitize_plugin_name(
    name: str,
    plugins_dir: Path,
    *,
    allow_subdir: bool = False,
) -> Path:
    """Validate a plugin name and return the safe target path inside *plugins_dir*.

    Raises ``ValueError`` on path-traversal sequences or a target outside the plugins directory.
    ``allow_subdir=True`` permits one forward slash so category-namespaced registry keys like
    ``observability/langfuse`` can be looked up; ``..`` and backslashes are still rejected.
    Install paths keep the default ``False`` because a fresh clone always lands top-level.
    """
    if allow_subdir and name:
        name = name.strip("/")
    if not name:
        raise ValueError("Plugin name must not be empty.")

    if name in {".", ".."}:
        raise ValueError(
            f"Invalid plugin name '{name}': must not reference the plugins directory itself."
        )

    # Reject obvious traversal characters
    bad_chars = ("\\", "..") if allow_subdir else ("/", "\\", "..")
    for bad in bad_chars:
        if bad in name:
            raise ValueError(f"Invalid plugin name '{name}': must not contain '{bad}'.")

    target = (plugins_dir / name).resolve()
    plugins_resolved = plugins_dir.resolve()

    if target == plugins_resolved:
        raise ValueError(
            f"Invalid plugin name '{name}': resolves to the plugins directory itself."
        )
    if plugins_resolved not in target.parents:
        raise ValueError(
            f"Invalid plugin name '{name}': resolves outside the plugins directory."
        )
    return target


_GITHUB_BROWSER_SEGMENTS = {
    "actions",
    "blob",
    "commit",
    "commits",
    "issues",
    "pull",
    "pulls",
    "releases",
    "tree",
    "wiki",
}


def _resolve_git_url(identifier: str) -> tuple[str, Optional[str]]:
    """Turn an identifier into a cloneable Git URL and optional subdirectory.

    NOTE: ``http://`` and ``file://`` schemes are accepted but will trigger a security warning at
    install time.
    """
    # Already a URL.
    if identifier.startswith(("https://", "http://", "git@", "ssh://", "file://")):
        if identifier.startswith("https://github.com/"):
            path = identifier[len("https://github.com/") :]
            path = path.split("?", 1)[0].split("#", 1)[0].strip("/")
            parts = path.split("/")
            if len(parts) >= 3 and all(parts[:2]) and parts[2] in _GITHUB_BROWSER_SEGMENTS:
                repo = parts[1].removesuffix(".git")
                subdir = None
                if parts[2] == "tree" and len(parts) >= 5:
                    subdir = "/".join(p for p in parts[4:] if p).strip("/") or None
                return f"https://github.com/{parts[0]}/{repo}.git", subdir

        # Explicit ``#subdir`` fragment — unambiguous for any scheme.
        if "#" in identifier:
            git_url, _, frag = identifier.partition("#")
            return git_url, (frag.strip("/") or None)
        # Natural ``.git/`` boundary (GitHub-style URLs).
        git_url, marker, subdir = identifier.partition(".git/")
        if marker:
            return git_url + ".git", (subdir.strip("/") or None)
        return identifier, None

    # owner/repo[/subdir...] shorthand
    parts = [p for p in identifier.strip("/").split("/") if p]
    if len(parts) >= 2:
        owner, repo = parts[0], parts[1]
        subdir = "/".join(parts[2:]).strip("/")
        git_url = f"https://github.com/{owner}/{repo}.git"
        return git_url, (subdir or None)

    raise ValueError(
        f"Invalid plugin identifier: '{identifier}'. "
        "Use a Git URL or 'owner/repo' shorthand (optionally with a subdirectory: "
        "'owner/repo/path/to/plugin')."
    )


def _resolve_subdir_within(clone_root: Path, subdir: str) -> Path:
    """Resolve ``subdir`` inside ``clone_root``, rejecting path traversal.

    Guards against ``..`` segments, absolute paths, and symlinks that would escape the clone.
    Raises ``PluginOperationError`` if the path escapes, doesn't exist, or is not a directory.
    """
    clone_root = clone_root.resolve()
    candidate = (clone_root / subdir).resolve()

    # The resolved candidate must stay within the clone root.
    if candidate != clone_root and clone_root not in candidate.parents:
        raise PluginOperationError(
            f"Plugin subdirectory '{subdir}' escapes the repository.",
        )

    if not candidate.exists():
        raise PluginOperationError(
            f"Plugin subdirectory '{subdir}' does not exist in the repository.",
        )
    if not candidate.is_dir():
        raise PluginOperationError(
            f"Plugin subdirectory '{subdir}' is not a directory.",
        )

    return candidate


def _repo_name_from_url(url: str) -> str:
    """Extract the repo name from a Git URL for the plugin directory name."""
    # Last path component after stripping trailing slashes and ``.git``; ssh-style
    # ``git@host:owner/repo`` has no slash to split on, hence the colon fallback.
    name = url.rstrip("/").removesuffix(".git").rsplit("/", 1)[-1]
    if ":" in name:
        name = name.rsplit(":", 1)[-1].rsplit("/", 1)[-1]
    return name


def _native_manifest_file(plugin_dir: Path) -> Optional[Path]:
    """``plugin.yaml`` (or ``plugin.yml``) under *plugin_dir*, or None when neither exists."""
    for name in ("plugin.yaml", "plugin.yml"):
        candidate = plugin_dir / name
        if candidate.exists():
            return candidate
    return None


def _has_portable_manifest(plugin_dir: Path) -> bool:
    """True when ``plugin.json`` exists (or is a symlink, even dangling) under *plugin_dir*."""
    portable_file = plugin_dir / "plugin.json"
    return portable_file.exists() or portable_file.is_symlink()


def _load_yaml_manifest(manifest_file: Path):
    """``yaml.safe_load`` of *manifest_file* (``{}`` when empty); raises on any read/parse error."""
    import yaml

    with open(manifest_file, encoding="utf-8") as f:
        return yaml.safe_load(f) or {}


def _read_portable_manifest(plugin_dir: Path) -> dict:
    """Validated Agent Plugins v1 ``plugin.json`` manifest (diagnostics dropped); raises on failure."""
    from hermes_cli.agent_plugins import read_agent_plugin_manifest

    manifest, _ = read_agent_plugin_manifest(plugin_dir)
    return manifest


def _read_manifest(plugin_dir: Path) -> dict:
    """Read a native or portable manifest, preferring native YAML."""
    manifest_file = _native_manifest_file(plugin_dir)
    if manifest_file is None:
        if not _has_portable_manifest(plugin_dir):
            return {}
        try:
            return _read_portable_manifest(plugin_dir)
        except Exception as e:
            logger.warning("Failed to read plugin.json in %s: %s", plugin_dir, e)
            return {}
    try:
        return _load_yaml_manifest(manifest_file)
    except Exception as e:
        logger.warning("Failed to read plugin.yaml in %s: %s", plugin_dir, e)
        return {}


def _looks_like_plugin_dir(target: Path) -> bool:
    """True when *target* has a native/portable manifest or a package ``__init__.py``."""
    return (
        _native_manifest_file(target) is not None
        or (target / "plugin.json").exists()
        or (target / "__init__.py").exists()
    )


def _copy_example_files(plugin_dir: Path, console) -> None:
    """Copy any .example files to their real names if they don't already exist.

    For example, ``config.yaml.example`` becomes ``config.yaml``. Skips files that already exist to
    avoid overwriting user config on reinstall.
    """
    for example_file in plugin_dir.glob("*.example"):
        real_name = example_file.stem  # e.g. "config.yaml" from "config.yaml.example"
        real_path = plugin_dir / real_name
        if real_path.exists():
            continue
        try:
            shutil.copy2(example_file, real_path)
            console.print(f"[dim]  Created {real_name} from {example_file.name}[/dim]")
        except OSError as e:
            console.print(f"[yellow]Warning:[/yellow] Failed to copy {example_file.name}: {e}")


def _requires_env_specs(manifest: dict) -> list[dict]:
    """Normalise ``requires_env`` (plain names or ``{name, description, url, secret}`` dicts)
    to a list of dicts; entries without a name are dropped."""
    env_specs: list[dict] = []
    for entry in manifest.get("requires_env") or []:
        if isinstance(entry, str):
            env_specs.append({"name": entry})
        elif isinstance(entry, dict) and entry.get("name"):
            env_specs.append(entry)
    return env_specs


def _missing_env_specs(manifest: dict) -> list[dict]:
    """Declared ``requires_env`` specs whose variable is unset in ``~/.hermes/.env``."""
    env_specs = _requires_env_specs(manifest)
    if not env_specs:
        return []

    from hermes_cli.config import get_env_value

    return [s for s in env_specs if not get_env_value(s["name"])]


def _missing_requires_env_names(manifest: dict) -> list[str]:
    """Return declared ``requires_env`` names that are unset in ``~/.hermes/.env``."""
    return [s["name"] for s in _missing_env_specs(manifest)]


def _print_python_dependencies(manifest: dict, console) -> None:
    """Surface declared python_dependencies at install time (#64165).

    Declaration seam ONLY — Hermes never auto-installs plugin pip dependencies (isolation design
    deferred; see #64165 / #15220). We print the declared requirements with a copy-pasteable install
    hint.
    """
    deps = manifest.get("python_dependencies") or []
    if not isinstance(deps, list):
        return
    deps = [d.strip() for d in deps if isinstance(d, str) and d.strip()]
    if not deps:
        return
    plugin_name = manifest.get("name", "this plugin")
    console.print(
        f"\n[bold]{plugin_name}[/bold] declares Python dependencies "
        "(not installed automatically):"
    )
    for dep in deps:
        console.print(f"  - {dep}")
    console.print(
        "[dim]Install them yourself if needed: "
        f"pip install {' '.join(repr(d) for d in deps)}[/dim]\n"
    )


def _prompt_plugin_env_vars(manifest: dict, console) -> None:
    """Prompt for required environment variables declared in plugin.yaml.

    ``requires_env`` accepts either a plain list of names or rich entries with ``name``,
    ``description``, ``url`` and ``secret``. Already-set variables are skipped; values are saved
    to the user's ``.env``.
    """
    missing = _missing_env_specs(manifest)
    if not missing:
        return

    from hermes_cli.config import save_env_value
    from hermes_constants import display_hermes_home

    plugin_name = manifest.get("name", "this plugin")
    console.print(f"\n[bold]{plugin_name}[/bold] requires the following environment variables:\n")

    for spec in missing:
        name = spec["name"]
        desc = spec.get("description", "")
        url = spec.get("url", "")
        secret = spec.get("secret", False)

        label = f"  {name}"
        if desc:
            label += f" — {desc}"
        console.print(label)
        if url:
            console.print(f"  [dim]Get yours at: {url}[/dim]")

        try:
            value = (masked_secret_prompt if secret else line_input)(f"  {name}: ").strip()
        except (EOFError, KeyboardInterrupt):
            console.print(f"\n[dim]  Skipped (you can set these later in {display_hermes_home()}/.env)[/dim]")
            return

        if value:
            save_env_value(name, value)
            os.environ[name] = value
            console.print(f"  [green]✓[/green] Saved to {display_hermes_home()}/.env")
        else:
            console.print(f"  [dim]  Skipped (set {name} in {display_hermes_home()}/.env later)[/dim]")

    console.print()


def _display_after_install(plugin_dir: Path, identifier: str) -> None:
    """Show after-install.md if it exists, otherwise a default message."""
    from rich.markdown import Markdown
    from rich.panel import Panel

    console = _console()
    after_install = plugin_dir / "after-install.md"

    if after_install.exists():
        body, title = Markdown(after_install.read_text(encoding="utf-8")), None
    else:
        body = f"[green bold]Plugin installed:[/] {identifier}\n[dim]Location:[/] {plugin_dir}"
        title = "✓ Installed"
    console.print()
    console.print(Panel(body, border_style="green", title=title, expand=False))
    console.print()


def _require_installed_plugin(name: str, plugins_dir: Path, console) -> Path:
    """Return the plugin path if it exists, or exit with an error (invalid name, or a listing of
    installed plugins when missing)."""
    try:
        target = _sanitize_plugin_name(name, plugins_dir, allow_subdir=True)
    except ValueError as e:
        console.print(f"[red]Error:[/red] {e}")
        sys.exit(1)
    if not target.exists():
        installed = ", ".join(d.name for d in plugins_dir.iterdir() if d.is_dir()) or "(none)"
        console.print(
            f"[red]Error:[/red] Plugin '{name}' not found in {plugins_dir}.\n"
            f"Installed plugins: {installed}"
        )
        sys.exit(1)
    return target


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------


_EXACT_COMMIT_RE = re.compile(r"^[0-9a-fA-F]{40}$")
_INSTALL_METADATA_FILE = ".install-metadata.json"

def _install_metadata_path() -> Path:
    return get_hermes_home() / "plugins" / _INSTALL_METADATA_FILE


def _read_install_metadata() -> dict[str, dict[str, object]]:
    """Read profile-local, non-secret plugin source metadata from disk."""
    path = _install_metadata_path()
    if not path.exists():
        return {}
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise PluginOperationError(f"Could not read plugin install metadata: {exc}") from exc
    if not isinstance(value, dict):
        raise PluginOperationError("Plugin install metadata must be a JSON object.")
    return value


def _write_install_metadata(metadata: dict[str, dict[str, object]]) -> None:
    """Atomically replace the profile-local plugin install metadata sidecar."""
    path = _install_metadata_path()
    atomic_write_text(
        path,
        json.dumps(metadata, indent=2, sort_keys=True) + "\n",
        tmp_prefix=f"{path.name}.tmp-",
    )


def _normalize_exact_revision(ref: str) -> str:
    if not isinstance(ref, str) or not _EXACT_COMMIT_RE.fullmatch(ref):
        raise PluginOperationError("--ref must be a full 40-character commit SHA.")
    return ref.lower()


def _safe_git_error(result: subprocess.CompletedProcess, source_url: str = "") -> str:
    """Return diagnosable Git output without echoing embedded credentials."""
    from agent.redact import redact_sensitive_text

    error = (result.stderr or result.stdout or "").strip()
    if source_url:
        error = error.replace(source_url, _scrub_git_url(source_url))
    return redact_sensitive_text(error)


def _git_or_raise(
    git_exe: str, repo: Path, *args: str, failure_prefix: str, timeout: int = 60, source_url: str = ""
) -> subprocess.CompletedProcess:
    """Run git in *repo*; on a non-zero exit raise PluginOperationError(prefix + scrubbed error)."""
    result = _run_plugin_git(git_exe, repo, *args, timeout=timeout)
    if result.returncode != 0:
        raise PluginOperationError(failure_prefix + _safe_git_error(result, source_url))
    return result


def _git_head_revision(repo: Path, git_exe: str) -> str:
    result = _git_or_raise(
        git_exe, repo, "rev-parse", "HEAD", timeout=15,
        failure_prefix="Could not determine installed Git revision:\n",
    )
    return result.stdout.strip().lower()


def _checkout_exact_revision(repo: Path, git_exe: str, revision: str) -> None:
    """Fetch and detach at one immutable commit, then verify the resulting HEAD."""
    steps = (
        (
            ("fetch", "--depth", "1", "origin", revision),
            f"Git fetch of commit '{revision}' timed out after 60 seconds.",
            f"Git commit '{revision}' could not be fetched:\n",
        ),
        (
            ("checkout", "--detach", revision),
            f"Git checkout of commit '{revision}' timed out after 60 seconds.",
            f"Git checkout of commit '{revision}' failed:\n",
        ),
    )
    for args, timeout_msg, failure_prefix in steps:
        try:
            _git_or_raise(git_exe, repo, *args, failure_prefix=failure_prefix)
        except subprocess.TimeoutExpired as exc:
            raise PluginOperationError(timeout_msg) from exc
    actual = _git_head_revision(repo, git_exe)
    if actual != revision:
        raise PluginOperationError(
            f"Checked-out revision '{actual}' does not match requested commit '{revision}'."
        )


def _scrub_git_url(git_url: str) -> str:
    """Strip credentials and query/fragment data from an HTTP Git URL."""
    parsed = urllib.parse.urlsplit(git_url)
    if parsed.scheme in {"http", "https"} and parsed.hostname:
        host = f"[{parsed.hostname}]" if ":" in parsed.hostname else parsed.hostname
        if parsed.port is not None:
            host = f"{host}:{parsed.port}"
        return urllib.parse.urlunsplit(
            (parsed.scheme, host, parsed.path, "", "")
        )
    return git_url


def _canonical_source(git_url: str, subdir: Optional[str]) -> str:
    scrubbed = _scrub_git_url(git_url)
    return f"{scrubbed}#{subdir}" if subdir else scrubbed


def _scrub_cloned_origin(repo: Path, git_exe: str, git_url: str) -> None:
    """Ensure credentials used for cloning do not survive in ``.git/config``."""
    scrubbed = _scrub_git_url(git_url)
    if scrubbed == git_url:
        return
    _git_or_raise(
        git_exe, repo, "remote", "set-url", "origin", scrubbed, timeout=15,
        failure_prefix="Could not sanitize installed Git remote:\n", source_url=git_url,
    )


def _check_manifest_version(manifest: dict, plugin_name: str) -> None:
    """Reject manifests declaring a newer ``manifest_version`` than this installer supports."""
    mv = manifest.get("manifest_version")
    if mv is None:
        return
    try:
        mv_int = int(mv)
    except (ValueError, TypeError):
        raise PluginOperationError(
            f"Plugin '{plugin_name}' has invalid manifest_version "
            f"'{mv}' (expected an integer).",
        ) from None
    if mv_int > _SUPPORTED_MANIFEST_VERSION:
        from hermes_cli.config import recommended_update_command

        raise PluginOperationError(
            f"Plugin '{plugin_name}' requires manifest_version {mv}, "
            f"but this installer only supports up to {_SUPPORTED_MANIFEST_VERSION}. "
            f"Run {recommended_update_command()} to update Hermes.",
        ) from None


def _clone_plugin_repo(tmp_clone: Path, git_url: str, revision: Optional[str]) -> str:
    """Shallow-clone *git_url* into *tmp_clone* (detached at *revision* when given), scrub any
    credentials from the recorded origin, and return the installed HEAD SHA."""
    git_exe = _resolve_git_executable()
    if not git_exe:
        raise PluginOperationError("git is not installed or not in PATH.")

    clone_args = ["clone", "--depth", "1"]
    if revision:
        clone_args.append("--no-checkout")
    clone_args.extend([git_url, str(tmp_clone)])
    try:
        result = _run_plugin_git(git_exe, tmp_clone.parent, *clone_args)
    except FileNotFoundError as e:
        raise PluginOperationError("git is not installed or not in PATH.") from e
    except subprocess.TimeoutExpired as e:
        raise PluginOperationError("Git clone timed out after 60 seconds.") from e
    if result.returncode != 0:
        err = _safe_git_error(result, git_url)
        raise PluginOperationError(f"Git clone failed:\n{err}")

    _scrub_cloned_origin(tmp_clone, git_exe, git_url)
    if revision:
        _checkout_exact_revision(tmp_clone, git_exe, revision)
    return _git_head_revision(tmp_clone, git_exe)


def _read_manifest_for_install(plugin_dir: Path) -> dict:
    """Manifest of a freshly cloned tree. Unlike :func:`_read_manifest`, a broken portable
    ``plugin.json`` is an install error (not a silent ``{}``) and its diagnostics are logged."""
    if _native_manifest_file(plugin_dir) is not None or not _has_portable_manifest(plugin_dir):
        return _read_manifest(plugin_dir)
    try:
        from hermes_cli.agent_plugins import read_agent_plugin_manifest

        manifest, diagnostics = read_agent_plugin_manifest(plugin_dir)
    except Exception as exc:
        raise PluginOperationError(
            f"Portable plugin manifest validation failed: {exc}"
        ) from exc
    for diagnostic in diagnostics:
        logger.warning("Agent Plugin install: %s", diagnostic.message)
    return manifest


def _swap_in_plugin(tmp_target: Path, target: Path, backup: Path, old_metadata: dict, new_metadata: dict) -> None:
    """Move the validated clone into place and persist metadata; on any failure restore the
    previous tree (if one was replaced) and the previous metadata sidecar, then re-raise."""
    replaced_existing = target.exists()
    if replaced_existing:
        os.replace(target, backup)
    try:
        os.replace(tmp_target, target)
        _write_install_metadata(new_metadata)
    except Exception:
        if target.exists():
            shutil.rmtree(target)
        if replaced_existing and backup.exists():
            os.replace(backup, target)
        if old_metadata:
            _write_install_metadata(old_metadata)
        else:
            _install_metadata_path().unlink(missing_ok=True)
        raise


def _install_plugin_core(
    identifier: str,
    *,
    force: bool,
    ref: Optional[str] = None,
    scan_decision_cb=None,
) -> tuple[Path, dict, str]:
    """Clone a Git plugin and atomically record its source and exact revision."""
    requested_revision = _normalize_exact_revision(ref) if ref is not None else None
    try:
        git_url, subdir = _resolve_git_url(identifier)
    except ValueError as e:
        raise PluginOperationError(str(e)) from e

    plugins_dir = _plugins_dir()
    source = _canonical_source(git_url, subdir)
    old_metadata = _read_install_metadata()

    # Reinstalling the same pinned source retains its pin, even if its plugin
    # directory was manually removed. Moving a pin requires an explicit --ref.
    if requested_revision is None:
        matching_pins = [
            entry
            for entry in old_metadata.values()
            if entry.get("source") == source and entry.get("pinned") is True
        ]
        if len(matching_pins) == 1 and isinstance(matching_pins[0].get("revision"), str):
            requested_revision = _normalize_exact_revision(matching_pins[0]["revision"])

    with tempfile.TemporaryDirectory(prefix=".install-", dir=plugins_dir) as tmp:
        tmp_clone = Path(tmp) / "plugin"
        installed_revision = _clone_plugin_repo(tmp_clone, git_url, requested_revision)
        tmp_target = (
            _resolve_subdir_within(tmp_clone, subdir) if subdir else tmp_clone
        )
        manifest = _read_manifest_for_install(tmp_target)
        plugin_name = manifest.get("name") or (
            subdir.rstrip("/").rsplit("/", 1)[-1] if subdir else _repo_name_from_url(git_url)
        )
        try:
            target = _sanitize_plugin_name(plugin_name, plugins_dir)
        except ValueError as e:
            raise PluginOperationError(str(e)) from e
        _check_manifest_version(manifest, plugin_name)

        # Security scan the clone BEFORE anything is moved into place
        # (see ``tools/plugin_guard.py``; inspired by Claude Cowork's skill
        # & plugin scanning). ``scan_decision_cb`` is called with the
        # ScanResult for caution verdicts and may return True to accept the
        # risk interactively. Raises PluginScanBlocked when blocked.
        _scan_plugin_tree(
            tmp_target,
            identifier,
            force=force,
            scan_decision_cb=scan_decision_cb,
        )

        if target.exists() and not force:
            raise PluginOperationError(
                f"Plugin '{plugin_name}' already exists. Use force reinstall "
                f"or run `hermes plugins update {plugin_name}`."
            )
        prior = old_metadata.get(plugin_name)
        if (
            target.exists()
            and requested_revision is None
            and isinstance(prior, dict)
            and prior.get("pinned") is True
        ):
            raise PluginOperationError(
                f"Plugin '{plugin_name}' is pinned. Reinstall it with an explicit "
                "--ref <40-character commit SHA> to change its source or revision."
            )

        new_metadata = dict(old_metadata)
        new_metadata[plugin_name] = {
            "pinned": requested_revision is not None,
            "revision": installed_revision,
            "source": source,
        }
        _swap_in_plugin(tmp_target, target, Path(tmp) / "previous-plugin", old_metadata, new_metadata)

    if not _looks_like_plugin_dir(target):
        logger.warning(
            "%s has no plugin.yaml / __init__.py; may not be a valid plugin",
            plugin_name,
        )

    _copy_example_files(target, _console())
    installed_manifest = _read_manifest(target)
    installed_name = installed_manifest.get("name") or target.name
    return target, installed_manifest, installed_name


def _looks_like_bare_index_name(identifier: str) -> bool:
    """True when *identifier* is a bare plugin name (no slash, not a URL).

    Bare names are resolved through the community plugin index; anything with a slash or URL scheme
    keeps the existing owner/repo / Git URL semantics.
    """
    if "/" in identifier or "\\" in identifier:
        return False
    return not identifier.startswith(("https://", "http://", "git@", "ssh://", "file://"))


def _resolve_index_name(identifier: str, console) -> tuple[str, Optional[str]]:
    """Resolve a bare plugin name to ``(install_identifier, pinned_ref)``.

    Exits with an error when the name is unknown, or lists candidates and exits when the name is
    ambiguous. The returned ref is only used when it is an exact 40-character commit SHA (the pin
    format the installer accepts); tag refs are surfaced as advisory output instead.
    """
    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}")
            console.print("Re-run with the exact name or the owner/repo identifier.")
        else:
            console.print(
                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."
            )
        sys.exit(1)

    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,
) -> None:
    """Install a plugin from a Git URL, owner/repo shorthand, or index name.

    Bare names (no slash, no URL scheme) are resolved through the community plugin index to
    ``owner/repo`` plus the index-pinned ref. An explicit ``--ref`` always wins over the index pin.

    After install, prompt "Enable now? [y/N]" unless *enable* is provided (True = auto-enable
    without prompting, False = install disabled).
    """
    console = _console()

    if _looks_like_bare_index_name(identifier):
        identifier, index_ref = _resolve_index_name(identifier, console)
        if ref is None:
            ref = index_ref

    try:
        git_url, _subdir = _resolve_git_url(identifier)
    except ValueError as e:
        console.print(f"[red]Error:[/red] {e}")
        sys.exit(1)

    if git_url.startswith(("http://", "file://")):
        console.print(
            "[yellow]Warning:[/yellow] Using insecure/local URL scheme. "
            "Consider using https:// or git@ for production installs.",
        )

    console.print(f"[dim]Cloning {git_url}{f' (subdir: {_subdir})' if _subdir else ''}...[/dim]")

    def _interactive_scan_decision(scan_result) -> bool:
        """Prompt the user to accept a caution-verdict plugin (Cowork 'warn')."""
        from tools.plugin_guard import format_scan_report

        console.print()
        console.print("[yellow]⚠ Security scan flagged this plugin:[/yellow]")
        console.print(format_scan_report(scan_result))
        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,
        )
    except PluginOperationError as e:
        label = "Blocked" if isinstance(e, PluginScanBlocked) else "Error"
        console.print(f"[red]{label}:[/red] {e}")
        sys.exit(1)

    if not _looks_like_plugin_dir(target):
        console.print(
            f"[yellow]Warning:[/yellow] {installed_name} doesn't contain plugin.yaml, "
            f"plugin.json, or __init__.py. It may not be a valid Hermes plugin.",
        )

    _prompt_plugin_env_vars(installed_manifest, console)

    _print_python_dependencies(installed_manifest, console)

    _display_after_install(target, identifier)

    should_enable = enable
    if should_enable is None:
        should_enable = _is_tty() and _ask_yes(f"  Enable '{installed_name}' now? [y/N]: ")

    if should_enable:
        _set_plugin_enabled(installed_name, enable=True)
        console.print(
            f"[green]✓[/green] Plugin [bold]{installed_name}[/bold] enabled.",
        )
    else:
        console.print(
            f"[dim]Plugin installed but not enabled. "
            f"Run `hermes plugins enable {installed_name}` to activate.[/dim]",
        )

    # Capability consent (#64228): if the manifest declares capabilities,
    # show the list once and record consent. Non-interactive installs (and
    # declines) proceed with capabilities ungranted — fail closed.
    declared_caps = _declared_capabilities_from_manifest(installed_manifest, installed_name)
    if declared_caps:
        _run_capability_consent(console, installed_name, declared_caps, context="install")

    console.print("[dim]Restart the gateway for the plugin to take effect:[/dim]")
    console.print("[dim]  hermes gateway restart[/dim]")
    console.print()


def _pull_plugin_update(target: Path, pinned_msg, not_git_msg, before_pull=None) -> str:
    """Shared ``update`` core: refuse pinned / non-git checkouts, ``git pull``, record the new
    revision. Returns the pull output; raises :class:`PluginOperationError` on any refusal.
    *pinned_msg(install_record)* / *not_git_msg()* build the caller-specific error text."""
    metadata = _read_install_metadata()
    install_record = metadata.get(target.name, {})
    if install_record.get("pinned") is True:
        raise PluginOperationError(pinned_msg(install_record))
    if not (target / ".git").exists():
        raise PluginOperationError(not_git_msg())
    if before_pull is not None:
        before_pull()
    ok, output = _git_pull_plugin_dir(target)
    if not ok:
        raise PluginOperationError(output)
    _record_pulled_revision(target, metadata, install_record)
    return output


def cmd_update(name: str) -> None:
    """Update an installed plugin by pulling latest from its git remote."""
    from rich.markup import escape

    console = _console()
    plugins_dir = _plugins_dir()
    target = _require_installed_plugin(name, plugins_dir, console)

    try:
        output = _pull_plugin_update(
            target,
            lambda rec: (
                f"Plugin '{name}' is pinned to {rec.get('revision')}. To move it, run "
                f"`hermes plugins install {escape(str(rec.get('source', '<source>')))} --force "
                "--ref <40-character commit SHA>`."
            ),
            lambda: (
                f"Plugin '{name}' was not installed from git (no .git directory). Cannot update."
            ),
            before_pull=lambda: console.print(f"[dim]Updating {name}...[/dim]"),
        )
    except PluginOperationError as exc:
        console.print(f"[red]Error:[/red] {exc}")
        sys.exit(1)

    # Re-scan after update — Cowork re-scans skills/plugins on edit, and an
    # update can introduce malicious content into a previously clean plugin.
    # The pull has already mutated the tree, so a dangerous verdict disables
    # the plugin rather than leaving it active.
    if _scan_on_install_enabled():
        from tools.plugin_guard import format_scan_report, scan_plugin, should_allow_plugin_install

        scan_result = scan_plugin(target, source=name)
        allowed, reason = should_allow_plugin_install(scan_result)
        if allowed is not True:
            console.print()
            console.print(f"[yellow]⚠ Security scan flagged the updated plugin:[/yellow] {reason}")
            console.print(format_scan_report(scan_result))
            if scan_result.verdict == "dangerous":
                if name in _get_enabled_set() or name not in _get_disabled_set():
                    _set_plugin_enabled(name, enable=False)
                console.print(
                    f"[red]Plugin '{name}' has been disabled.[/red] Review the "
                    f"findings, then re-enable with `hermes plugins enable {name}` "
                    f"if you trust them.",
                )

    _post_pull_housekeeping(target, console)

    # Update-time re-consent (#64228): if the new version declares
    # capabilities the granted set lacks, surface the diff and require
    # re-consent for the additions. The stored consent hash detects a
    # changed declaration; additions stay ungranted until the user says yes
    # (non-interactive updates leave them ungranted — fail closed).
    updated_manifest = _read_manifest(target)
    plugin_id = updated_manifest.get("name") or target.name
    declared_caps = _declared_capabilities_from_manifest(updated_manifest, plugin_id)
    if declared_caps:
        from hermes_cli.plugin_capabilities import declared_set_changed, pending_capabilities

        if pending_capabilities(plugin_id, declared_caps) or declared_set_changed(
            plugin_id, declared_caps
        ):
            _run_capability_consent(console, plugin_id, declared_caps, context="update")

    out = output.strip()
    if "Already up to date" in out:
        console.print(
            f"[green]✓[/green] Plugin [bold]{name}[/bold] is already up to date."
        )
    else:
        console.print(f"[green]✓[/green] Plugin [bold]{name}[/bold] updated.")
        console.print(f"[dim]{out}[/dim]")


def _post_pull_housekeeping(target: Path, console) -> None:
    """After ``git pull``: drop ``__pycache__`` compiled from the previous revision (same stale-
    bytecode class as the main checkout, #6207/#60242) and copy any new ``.example`` files."""
    _clear_plugin_bytecode(target)
    _copy_example_files(target, console)


def _record_pulled_revision(target: Path, metadata: dict, install_record: dict) -> None:
    """After a pull, store the new HEAD in the plugin's install-metadata record (if it has one)."""
    if not install_record:
        return
    git_exe = _resolve_git_executable()
    if git_exe:
        install_record["revision"] = _git_head_revision(target, git_exe)
        metadata[target.name] = install_record
        _write_install_metadata(metadata)


def _remove_plugin_core(target: Path) -> None:
    """Remove one plugin and its metadata without splitting their state."""
    metadata = _read_install_metadata()
    if target.name not in metadata:
        shutil.rmtree(target)
        return

    updated = dict(metadata)
    updated.pop(target.name)
    staging = Path(
        tempfile.mkdtemp(prefix=f".{target.name}.remove-", dir=target.parent)
    )
    backup = staging / "plugin"
    os.replace(target, backup)
    try:
        _write_install_metadata(updated)
    except Exception:
        try:
            os.replace(backup, target)
        except OSError as restore_exc:
            raise PluginOperationError(
                f"Plugin metadata update failed and '{target.name}' could not be "
                f"restored automatically; recovery copy remains at {backup}."
            ) from restore_exc
        shutil.rmtree(staging, ignore_errors=True)
        raise
    shutil.rmtree(staging)


def cmd_remove(name: str) -> None:
    """Remove an installed plugin by name."""
    console = _console()
    plugins_dir = _plugins_dir()
    target = _require_installed_plugin(name, plugins_dir, console)

    try:
        _remove_plugin_core(target)
    except (OSError, PluginOperationError) as exc:
        console.print(f"[red]Error:[/red] Could not remove plugin '{name}': {exc}")
        sys.exit(1)
    console.print()
    console.print(f"[red]✗[/red] Plugin [bold]{name}[/bold] removed from {plugins_dir}")
    console.print()


def _get_disabled_set() -> set:
    """Read the disabled plugins set from config.yaml.

    An explicit deny-list. A plugin name here never loads, even if also listed in
    ``plugins.enabled``.
    """
    return _config_name_set("plugins", "disabled")


def _save_disabled_set(disabled: set) -> None:
    """Write the disabled plugins list to config.yaml."""
    _write_config_value("plugins", "disabled", sorted(disabled))


_BASIC_AUTH_PLUGIN_KEYS = frozenset({"basic", "dashboard_auth/basic"})


def ensure_basic_auth_plugin_enabled_in_config(cfg: dict) -> bool:
    """Re-enable the bundled basic dashboard-auth plugin in *cfg*.

    ``hermes setup`` / ``hermes plugins disable basic`` can park the plugin in
    ``plugins.disabled`` while ``dashboard.basic_auth`` is configured; the bundled provider
    still respects the deny-list, so password auth silently fails until the block is removed.
    Returns True when modified.
    """
    plugins_cfg = cfg.get("plugins")
    disabled = plugins_cfg.get("disabled") if isinstance(plugins_cfg, dict) else None
    if not isinstance(disabled, list) or not (set(disabled) & _BASIC_AUTH_PLUGIN_KEYS):
        return False
    plugins_cfg["disabled"] = sorted(set(disabled) - _BASIC_AUTH_PLUGIN_KEYS)
    return True


def _get_enabled_set() -> set:
    """Read the enabled plugins allow-list from config.yaml.

    Plugins are opt-in: only names here are loaded. Returns ``set()`` if the key is missing (same
    behaviour as "nothing enabled yet").
    """
    return _config_name_set("plugins", "enabled")


def _save_enabled_set(enabled: set) -> None:
    """Write the enabled plugins list to config.yaml."""
    _write_config_value("plugins", "enabled", sorted(enabled))


def _discard_key_and_leaf(names: set, key: str) -> None:
    """Drop *key* and its bare leaf (``observability/langfuse`` -> ``langfuse``) from *names*, so a
    stale legacy bare-name entry can't keep vetoing the canonical key."""
    names.discard(key)
    names.discard(key.split("/")[-1])


def _set_plugin_enabled(name: str, *, enable: bool) -> None:
    """Move *name* between the enabled allow-list and the disabled deny-list and persist both."""
    enabled = _get_enabled_set()
    disabled = _get_disabled_set()
    (enabled.add if enable else enabled.discard)(name)
    (disabled.discard if enable else disabled.add)(name)
    _save_enabled_set(enabled)
    _save_disabled_set(disabled)


def _resolve_plugin_key(name: str) -> Optional[str]:
    """Resolve a user-supplied plugin identifier to its canonical registry key.

    Accepts the bare manifest name, the directory name, or the path-derived key
    (``observability/langfuse``) and returns the key the loader gates on; ``None`` when nothing
    matches. Single normalization point so ``enable``/``disable`` write the same key that
    ``PluginManager`` matches against, nested category plugins included.
    """
    resolved = _resolve_plugin_key_and_source(name)
    return resolved[0] if resolved else None


def _find_plugin_entry(name: str) -> Optional[tuple]:
    """First discovered ``(name, version, description, source, dir_path, key)`` entry whose
    manifest name or canonical key equals *name*."""
    return next(
        (entry for entry in _discover_all_plugins() if name in (entry[0], entry[5])), None
    )


def _resolve_plugin_key_and_source(name: str) -> Optional[tuple]:
    """Resolve *name* to ``(canonical_key, source)`` or ``None`` if no match.

    Same normalization as :func:`_resolve_plugin_key` but also returns the plugin's source
    (``"bundled"``, ``"user"``, ``"project"``, ...) so the enable path can tell whether a built-
    in-override consent prompt is needed.
    """
    entries = _discover_all_plugins()
    # 1. Exact match on canonical key or manifest name — always unambiguous.
    for entry in entries:
        if name in (entry[0], entry[5]):
            return (entry[5], entry[3])
    # 2. Fall back to a bare leaf-name match (e.g. "langfuse" ->
    #    "observability/langfuse"), but only when it resolves to exactly one
    #    plugin so we never silently pick the wrong same-named nested plugin.
    leaf_matches = [
        (entry[5], entry[3]) for entry in entries
        if name == entry[5].split("/")[-1]
    ]
    if len(leaf_matches) == 1:
        return leaf_matches[0]
    return None


def _set_plugin_entry_flag(plugin_id: str, key: str, value: bool) -> None:
    """Write ``plugins.entries.<plugin_id>.<key> = value`` into config.yaml."""
    from hermes_cli.config import load_config, save_config
    config = load_config()
    entry = _sub_dict(_sub_dict(_sub_dict(config, "plugins"), "entries"), plugin_id)
    entry[key] = bool(value)
    save_config(config)


def cmd_enable(name: str, allow_tool_override: Optional[bool] = None) -> None:
    """Add a plugin to the enabled allow-list (and remove it from disabled).

    Non-bundled plugins are prompted about the privileged ``allow_tool_override`` capability
    (replacing built-in tools). ``allow_tool_override`` is tri-state: ``True`` grants and
    ``False`` declines without prompting, ``None`` asks interactively. Bundled plugins are
    trusted and never prompted.
    """
    from hermes_cli.relay_plugin_cutover import (
        LEGACY_RELAY_PLUGIN_KEYS,
        RELAY_PLUGINS_CONFIG_ENV,
    )

    console = _console()

    def _refuse_legacy_relay(plugin: str) -> None:
        if plugin in LEGACY_RELAY_PLUGIN_KEYS:
            console.print(
                f"[red]Plugin '{plugin}' was removed.[/red] Relay lifecycle is owned "
                f"by Hermes core; configure {RELAY_PLUGINS_CONFIG_ENV} instead."
            )
            sys.exit(1)

    _refuse_legacy_relay(name)

    # Discover the plugin — check installed (user) AND bundled, including
    # nested category plugins — and normalize to its canonical registry key.
    resolved = _resolve_plugin_key_and_source(name)
    if resolved is None:
        console.print(f"[red]Plugin '{name}' is not installed or bundled.[/red]")
        sys.exit(1)
    key, source = resolved
    _refuse_legacy_relay(key)

    enabled = _get_enabled_set()
    disabled = _get_disabled_set()

    already_enabled = key in enabled and key not in disabled

    if not already_enabled:
        enabled.add(key)
        # Drop every alias of this plugin from the disabled list so an
        # explicit disable under a different form can't keep it off. The
        # loader's disable check matches on BOTH the canonical key
        # (``web/firecrawl``) AND the manifest name (``web-firecrawl``);
        # a stale entry under either form makes "explicit disable wins"
        # (plugins.py) silently veto this enable. Discard the key, its
        # bare leaf, and the manifest name. (#40190 follow-up.)
        _discard_key_and_leaf(disabled, key)
        for entry in _discover_all_plugins():
            # entry = (name, version, description, source, dir_path, key)
            if entry[5] == key:
                disabled.discard(entry[0])
                break
        _save_enabled_set(enabled)
        _save_disabled_set(disabled)
        console.print(
            f"[green]✓[/green] Plugin [bold]{key}[/bold] enabled. "
            "Takes effect on next session."
        )
    else:
        console.print(f"[dim]Plugin '{key}' is already enabled.[/dim]")

    # Built-in tool override is a privileged grant. Bundled plugins ship with
    # Hermes core and are trusted; every other source needs operator opt-in.
    if source == "bundled":
        return

    # Capability consent (#64228): when the manifest declares capabilities,
    # the consent screen is the canonical grant path — it covers
    # tools.override too, so skip the legacy standalone prompt unless the
    # operator explicitly passed --allow-tool-override/--no-allow-tool-override.
    declared_caps = _declared_capabilities_for_key(key)
    if declared_caps:
        _run_capability_consent(console, key, declared_caps, context="enable")
        if allow_tool_override is not None:
            _resolve_tool_override_grant(console, key, allow_tool_override)
        return

    _resolve_tool_override_grant(console, key, allow_tool_override)


# ── Capability consent flow (#64228) ─────────────────────────────────────────


def _declared_capabilities_from_manifest(manifest: dict, plugin_name: str = "?") -> list:
    """Extract + normalize the ``capabilities:`` declaration from a manifest."""
    from hermes_cli.plugin_capabilities import parse_declared_capabilities

    return parse_declared_capabilities(
        (manifest or {}).get("capabilities"), plugin_name
    )


def _declared_capabilities_for_key(key: str) -> list:
    """Read the declared capabilities for an installed/bundled plugin by key."""
    entry = _find_plugin_entry(key)
    if entry is None:
        return []
    if entry[3] == "entrypoint":
        from hermes_cli.plugins import discover_entrypoint_manifests

        for manifest in discover_entrypoint_manifests():
            if key in (manifest.key, manifest.name):
                return list(manifest.capabilities)
        return []
    dir_path = entry[4]
    if not dir_path:
        return []
    return _declared_capabilities_from_manifest(_read_manifest(Path(dir_path)), entry[0])


def _print_capability_list(console, capabilities: list) -> None:
    """Render the consent screen body: one line per capability."""
    from hermes_cli.plugin_capabilities import CAPABILITY_REGISTRY

    for cap in capabilities:
        spec = CAPABILITY_REGISTRY.get(cap)
        desc = spec.description if spec else ""
        console.print(f"    [bold]{cap}[/bold] — {desc}")


def _run_capability_consent(
    console,
    plugin_id: str,
    declared: list,
    *,
    context: str = "install",
) -> bool:
    """Show the capability consent screen and record the decision.

    Lists declared capabilities with risk descriptions and asks one Y/n. On consent the pending
    capabilities are granted under ``plugins.entries.<id>.granted_capabilities`` with a hash of
    the declared set. On decline — or in ANY non-interactive context — they stay ungranted (fail
    closed) and the plugin must degrade via ``ctx.has_capability()``. This is consent + audit,
    NOT a sandbox: an in-process plugin can run arbitrary Python regardless. Returns True when
    granted.
    """
    from hermes_cli.plugin_capabilities import (
        pending_capabilities,
        record_consent,
    )

    pending = pending_capabilities(plugin_id, declared)
    if not pending:
        # Everything declared is already granted — refresh the consent hash
        # so a later declaration change is detected against the current set.
        if declared:
            record_consent(plugin_id, [], declared)
        return True

    verb = "requests" if context == "install" else "now requests"
    console.print(
        f"\n  [yellow]Plugin [bold]{plugin_id}[/bold] {verb} the following "
        "capabilities:[/yellow]"
    )
    _print_capability_list(console, pending)
    console.print(
        "  [dim]Granting trusts the plugin author with these host surfaces. "
        "This is consent, not a sandbox — plugins run as regular Python "
        "in-process.[/dim]"
    )

    if not _is_tty():
        console.print(
            "  [yellow]Non-interactive session: capabilities NOT granted "
            "(fail closed).[/yellow] Run "
            f"`hermes plugins capabilities {plugin_id}` to review and "
            f"`hermes plugins enable {plugin_id}` to grant interactively."
        )
        return False

    if _ask_yes("  Grant these capabilities? [y/N] ", console.input):
        record_consent(plugin_id, pending, declared)
        console.print(
            f"  [green]✓[/green] Granted: {', '.join(pending)} "
            f"([dim]plugins.entries.{plugin_id}.granted_capabilities[/dim])"
        )
        return True

    console.print(
        f"  [dim]Declined. {plugin_id} stays enabled with these capabilities "
        "off; it should degrade gracefully (ctx.has_capability()). Re-run "
        f"`hermes plugins enable {plugin_id}` to grant later.[/dim]"
    )
    return False


def cmd_capabilities(name: Optional[str] = None) -> None:
    """``hermes plugins capabilities [<id>]`` — declared vs granted."""
    from hermes_cli.plugin_capabilities import (
        CAPABILITY_REGISTRY,
        granted_capabilities,
        plugin_capability_granted,
    )

    console = _console()

    rows = []
    for entry in _discover_all_plugins():
        # entry = (name, version, description, source, dir_path, key)
        key = entry[5] or entry[0]
        if name is not None and name not in (key, entry[0]):
            continue
        declared = _declared_capabilities_for_key(key)
        granted = granted_capabilities(key)
        # Legacy grants surface too: report capabilities live via deprecated
        # allow_* keys so `capabilities` shows the true effective state.
        effective = {
            cap for cap in CAPABILITY_REGISTRY
            if plugin_capability_granted(key, cap)
        }
        if not declared and not effective and name is None:
            continue
        rows.append((key, entry[3], declared, granted, effective))

    if name is not None and not rows:
        console.print(f"[red]Plugin '{name}' is not installed or bundled.[/red]")
        sys.exit(1)

    if not rows:
        console.print("[dim]No plugins declare or hold capabilities.[/dim]")
        return

    for key, source, declared, granted, effective in sorted(rows):
        console.print(f"[bold]{key}[/bold] [dim]({source})[/dim]")
        if not declared:
            console.print("  declared: [dim](none)[/dim]")
        for cap in declared:
            if cap in effective:
                mark = "[green]granted[/green]"
                if cap not in granted:
                    mark += " [dim](via legacy allow_* key — deprecated)[/dim]"
            else:
                mark = "[yellow]not granted[/yellow]"
            console.print(f"  {cap}: {mark}")
        for cap in sorted(effective - set(declared)):
            console.print(
                f"  {cap}: [green]granted[/green] "
                "[dim](not declared in manifest)[/dim]"
            )


def _resolve_tool_override_grant(
    console, key: str, allow_tool_override: Optional[bool]
) -> None:
    """Resolve and persist the ``allow_tool_override`` grant for a plugin."""
    if allow_tool_override is None:
        # Interactive consent. Default to NO so a blind Enter doesn't grant
        # a privileged capability, and a non-interactive stdin denies safely.
        prompt = (
            "[yellow]Allow this plugin to replace built-in tools "
            "(e.g. shell_exec, write_file)?[/yellow]\n"
            "  This is a privileged capability: an override can intercept "
            "everything the agent routes through that tool.\n"
            "  Grant it? [y/N] "
        )
        allow_tool_override = _ask_yes(prompt, console.input)

    _set_plugin_entry_flag(key, "allow_tool_override", allow_tool_override)
    if allow_tool_override:
        console.print(
            f"[green]✓[/green] Granted [bold]{key}[/bold] permission to "
            "override built-in tools "
            f"([dim]plugins.entries.{key}.allow_tool_override: true[/dim])."
        )
    else:
        console.print(
            f"[dim]{key} may not override built-in tools. Re-run "
            f"`hermes plugins enable {key} --allow-tool-override` to grant "
            "this later.[/dim]"
        )


def cmd_disable(name: str) -> None:
    """Remove a plugin from the enabled allow-list (and add to disabled)."""
    console = _console()
    key = _resolve_plugin_key(name)
    if key is None:
        console.print(f"[red]Plugin '{name}' is not installed or bundled.[/red]")
        sys.exit(1)

    enabled = _get_enabled_set()
    disabled = _get_disabled_set()

    if key not in enabled and key in disabled:
        console.print(f"[dim]Plugin '{key}' is already disabled.[/dim]")
        return

    # Drop any legacy bare-name entry from the allow-list too, so a stale
    # bare name can't keep a nested plugin loading after an explicit disable.
    _discard_key_and_leaf(enabled, key)
    disabled.add(key)
    _save_enabled_set(enabled)
    _save_disabled_set(disabled)
    console.print(
        f"[yellow]\u2298[/yellow] Plugin [bold]{key}[/bold] disabled. "
        "Takes effect on next session."
    )


def _read_manifest_info(d: Path, prefix: str):
    """Read a native or portable manifest and return display metadata."""
    manifest_file = _native_manifest_file(d)
    if manifest_file is None:
        if not _has_portable_manifest(d):
            return None
        try:
            manifest = _read_portable_manifest(d)
            name = manifest["name"]
        except Exception:
            return None
    else:
        # Unreadable YAML (or no yaml module) degrades to the directory name, silently.
        try:
            manifest = _load_yaml_manifest(manifest_file)
        except Exception:
            manifest = {}
        manifest = manifest if isinstance(manifest, dict) else {}
        name = manifest.get("name", d.name)
    key = f"{prefix}/{d.name}" if prefix else name
    return name, manifest.get("version", ""), manifest.get("description", ""), key


def _is_portable_plugin_dir(dir_path) -> bool:
    """True when *dir_path* is an Agent Plugins v1 package (``plugin.json``
    only — a native ``plugin.yaml`` takes precedence, matching the loader)."""
    try:
        d = Path(dir_path)
        return d.is_dir() and _native_manifest_file(d) is None and _has_portable_manifest(d)
    except OSError:
        return False


# Manifest kinds that are active-by-default when bundled: backends auto-load,
# platforms register lazily but are available out of the box, model providers
# run through providers/ discovery (see PluginManager.discover_and_load).
_BUNDLED_DEFAULT_ON_KINDS = frozenset({"backend", "platform", "model-provider"})


def _bundled_default_on(dir_path) -> bool:
    """True when a bundled plugin at *dir_path* is active without an explicit
    ``plugins.enabled`` entry. Standalone/exclusive kinds stay opt-in, and
    portable packages (``plugin.json``) have no kind at all."""
    manifest_file = _native_manifest_file(Path(dir_path))
    if manifest_file is None:
        return False
    try:
        kind = str(_load_yaml_manifest(manifest_file).get("kind", "standalone")).strip().lower()
        return kind in _BUNDLED_DEFAULT_ON_KINDS
    except Exception:
        return False


def _scan_level(
    base: Path,
    source: str,
    skip_names: set,
    prefix: str,
    depth: int,
    seen: dict,
) -> None:
    """Recursive directory scan matching PluginManager._scan_directory_level."""
    if not base.is_dir():
        return
    for d in sorted(base.iterdir()):
        if not d.is_dir():
            continue
        if depth == 0 and skip_names and d.name in skip_names:
            continue
        info = _read_manifest_info(d, prefix)
        if info is None:
            if depth == 0:
                _scan_level(d, source, set(), f"{prefix}/{d.name}" if prefix else d.name, 1, seen)
            continue
        name, version, description, key = info
        if key in seen and source == "bundled":
            continue
        src_label = "git" if source == "user" and (d / ".git").exists() else source
        seen[key] = (name, version, description, src_label, d, key)


def _discover_all_plugins() -> list:
    """Return (name, version, description, source, dir_path, key) for every plugin the loader sees.

    Matches the ordering/dedup of ``PluginManager.discover_and_load``: bundled, then user, then
    project, then entry points; later sources override earlier ones on key collision.
    """
    seen: dict = {}  # key -> (name, version, description, source, path, key)

    # Bundled (<repo>/plugins/<name>/), excluding memory/, context_engine/
    # and model-providers/ — model providers load through the dedicated
    # provider registry (providers/__init__.py), not the general PluginManager
    # opt-in surface, so listing them as toggleable plugins is misleading.
    from hermes_cli.plugins import get_bundled_plugins_dir
    repo_plugins = get_bundled_plugins_dir()
    for base, source, skip in (
        (repo_plugins, "bundled", {"memory", "context_engine", "model-providers"}),
        (_plugins_dir(), "user", set()),
    ):
        _scan_level(base, source, skip, "", 0, seen)

    # Entry-point plugins (installed as Python packages; no plugin directory).
    for name, version, description, path in _discover_entrypoint_plugins():
        seen[name] = (name, version, description, "entrypoint", path, name)
    return list(seen.values())


def _discover_entrypoint_plugins() -> list[tuple[str, str, str, str]]:
    """Return plugin entries advertised through ``hermes_agent.plugins``.

    Entry-point plugins are installed as Python packages, so they do not have a plugin directory
    under ``~/.hermes/plugins``. Include package metadata here so ``hermes plugins list`` can show
    and enable them.
    """
    from hermes_cli.plugins import ENTRY_POINTS_GROUP

    try:
        eps = importlib.metadata.entry_points()
        if hasattr(eps, "select"):
            group_eps = eps.select(group=ENTRY_POINTS_GROUP)
        elif isinstance(eps, dict):
            group_eps = eps.get(ENTRY_POINTS_GROUP, [])
        else:
            group_eps = [ep for ep in eps if ep.group == ENTRY_POINTS_GROUP]
    except Exception as exc:
        logger.debug("Entry-point plugin discovery failed: %s", exc)
        return []

    entries: list[tuple[str, str, str, str]] = []
    for ep in group_eps:
        dist = getattr(ep, "dist", None)
        metadata = getattr(dist, "metadata", None)
        if metadata is None:
            entries.append((ep.name, "", "", ep.value))
        else:
            entries.append((
                ep.name,
                str(getattr(dist, "version", "") or ""),
                str(metadata.get("Summary", "") or ""),
                ep.value,
            ))
    return entries


def _plugin_status(name: str, enabled: set, disabled: set, key: str = "") -> str:
    """Return the user-facing activation state for a plugin name or key."""
    if name in disabled or key in disabled:
        return "disabled"
    if name in enabled or key in enabled:
        return "enabled"
    return "not enabled"


def _filter_plugin_entries(entries: list, args: Any, enabled: set, disabled: set) -> list:
    """Apply ``hermes plugins list`` CLI filters."""
    filtered = entries
    if getattr(args, "no_bundled", False) or getattr(args, "user", False):
        filtered = [entry for entry in filtered if entry[3] != "bundled"]
    if getattr(args, "enabled", False):
        filtered = [
            entry for entry in filtered
            if _plugin_status(entry[0], enabled, disabled, key=entry[5]) == "enabled"
        ]
    return filtered


_STATUS_MARKUP = {"disabled": "[red]disabled[/red]", "enabled": "[green]enabled[/green]"}


def cmd_list(args: Any | None = None) -> None:
    """List all plugins (bundled + user) with enabled/disabled state."""
    from rich.table import Table

    console = _console()
    entries = _discover_all_plugins()
    if not entries:
        console.print("[dim]No plugins installed.[/dim]")
        console.print("[dim]Install with:[/dim] hermes plugins install owner/repo")
        return

    enabled = _get_enabled_set()
    disabled = _get_disabled_set()
    entries = _filter_plugin_entries(entries, args, enabled, disabled)

    # (name, status, version, description, source) per entry
    rows = [
        (name, _plugin_status(name, enabled, disabled, key=key), str(version), description, source)
        for name, version, description, source, _dir, key in entries
    ]

    if getattr(args, "json", False):
        keys = ("name", "status", "version", "description", "source")
        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:
            print(f"{status:12} {source:8} {version:8} {name}")
        return

    if not entries:
        console.print("[dim]No plugins matched the selected filters.[/dim]")
        return

    table = Table(title="Plugins", show_lines=False)
    table.add_column("Name", style="bold")
    table.add_column("Status")
    table.add_column("Version", style="dim")
    table.add_column("Description")
    table.add_column("Source", style="dim")

    for name, status_name, version, description, source in rows:
        status = _STATUS_MARKUP.get(status_name, "[yellow]not enabled[/yellow]")
        table.add_row(name, status, version, description, source)

    console.print()
    console.print(table)
    console.print()
    console.print("[dim]Compact view:[/dim] hermes plugins list --plain --no-bundled")
    console.print("[dim]Interactive toggle:[/dim] hermes plugins")
    console.print("[dim]Enable/disable:[/dim] hermes plugins enable/disable <name>")
    console.print("[dim]Plugins are opt-in by default — only 'enabled' plugins load.[/dim]")


# ---------------------------------------------------------------------------
# Provider plugin discovery helpers
# ---------------------------------------------------------------------------


def _discover_memory_providers() -> list[tuple[str, str]]:
    """Return [(name, description), ...] for available memory providers."""
    try:
        from plugins.memory import discover_memory_providers
        return [(name, desc) for name, desc, _avail in discover_memory_providers()]
    except Exception:
        return []


def _discover_context_engines() -> list[tuple[str, str]]:
    """Return [(name, description), ...] for available context engines.

    Includes repo-shipped engines from ``plugins/context_engine/`` AND plugin-registered engines
    (via ``ctx.register_context_engine``). Repo-shipped descriptions win on a name collision.
    """
    engines: dict[str, str] = {}

    try:
        from plugins.context_engine import discover_context_engines
        for name, desc, _avail in discover_context_engines():
            engines.setdefault(name, desc)
    except Exception:
        pass

    try:
        from hermes_cli.plugins import discover_plugins, get_plugin_context_engine
        discover_plugins()
        plugin_engine = get_plugin_context_engine()
        if plugin_engine and getattr(plugin_engine, "name", None):
            engines.setdefault(plugin_engine.name, "installed plugin")
    except Exception:
        pass

    return list(engines.items())


# memory.provider ("" = built-in) and context.engine config accessors.
_get_current_memory_provider = functools.partial(_config_str, "memory", "provider", default="")
_get_current_context_engine = functools.partial(_config_str, "context", "engine", default="compressor")
_save_memory_provider = functools.partial(_write_config_value, "memory", "provider")
_save_context_engine = functools.partial(_write_config_value, "context", "engine")


def _configure_provider_category(
    title: str, default_label: str, default_name: str, current: str, choices, save
) -> bool:
    """Radio picker: the built-in default first, then *choices*; a current value not among
    them is appended as ``(not found)``. Calls *save* and returns True when the choice changed."""
    from hermes_cli.curses_ui import curses_radiolist

    names = [default_name] + [name for name, _desc in choices]
    items = [default_label] + [f"{name} \u2014 {desc}" if desc else name for name, desc in choices]
    if current not in names:
        names.append(current)
        items.append(f"{current} (not found)")
    selected = max(i for i, name in enumerate(names) if name == current)

    new_value = names[curses_radiolist(title=title, items=items, selected=selected)]
    if new_value != current:
        save(new_value)
        return True
    return False


# (title, default label, default name, current-value reader, discovery fn, saver) per provider
# category. Readers/savers are looked up at call time so module-level patching still applies.
_PROVIDER_CATEGORY_SPECS = (
    ("Memory Provider", "built-in", "", lambda: _get_current_memory_provider(),
     lambda: _discover_memory_providers(), lambda v: _save_memory_provider(v)),
    ("Context Engine", "compressor", "compressor", lambda: _get_current_context_engine(),
     lambda: _discover_context_engines(), lambda v: _save_context_engine(v)),
)


def _configure_category_spec(spec) -> bool:
    """Launch the radio picker for one ``_PROVIDER_CATEGORY_SPECS`` row. Returns True if changed."""
    title, default_label, default_name, current, discover, save = spec
    return _configure_provider_category(
        f"{title} (select one)", f"{default_label} (default)", default_name,
        current(), discover(), save,
    )


def _provider_categories() -> list:
    """``[(title, current_label, configure_fn), ...]`` rows for the composite UI."""
    return [
        (spec[0], spec[3]() or spec[1], functools.partial(_configure_category_spec, spec))
        for spec in _PROVIDER_CATEGORY_SPECS
    ]


# ---------------------------------------------------------------------------
# Composite plugins UI
# ---------------------------------------------------------------------------


def cmd_show(name: str) -> None:
    """Show details for a single plugin, including declared emits/listens."""
    console = _console()
    match = _find_plugin_entry(name)
    if match is None:
        console.print(f"[red]Plugin '{name}' not found.[/red]")
        console.print("[dim]List installed plugins:[/dim] hermes plugins list")
        sys.exit(1)

    pname, version, description, source, dir_path, key = match
    manifest = _read_manifest(Path(dir_path)) if dir_path else {}
    emits = manifest.get("emits") or []
    listens = manifest.get("listens") or []

    enabled = _get_enabled_set()
    disabled = _get_disabled_set()
    status = _plugin_status(pname, enabled, disabled, key=key)

    console.print()
    console.print(f"[bold]{pname}[/bold]" + (f" [dim]v{version}[/dim]" if version else ""))
    if description:
        console.print(description)
    console.print(f"[dim]Status:[/dim] {status}")
    console.print(f"[dim]Source:[/dim] {source}")
    console.print(f"[dim]Key:[/dim] {key}")
    console.print(
        "[dim]Emits:[/dim] " + (", ".join(emits) if emits else "[dim](none)[/dim]")
    )
    console.print(
        "[dim]Listens:[/dim] " + (", ".join(listens) if listens else "[dim](none)[/dim]")
    )
    console.print()


def cmd_toggle() -> None:
    """Interactive composite UI — general plugins + provider plugin categories."""
    console = _console()

    # -- General plugins discovery (bundled + user) --
    entries = _discover_all_plugins()
    enabled_set = _get_enabled_set()
    disabled_set = _get_disabled_set()

    # Track by CANONICAL KEY (``key``), not the manifest name. The loader
    # (PluginManager) and ``cmd_enable``/``cmd_disable`` all gate on the
    # canonical key (``web/firecrawl``), while the manifest name may differ
    # (``web-firecrawl``). Persisting the bare name here caused the two
    # forms to drift: the menu would write ``web-firecrawl`` to
    # plugins.disabled, but ``hermes plugins enable web/firecrawl`` cleared
    # only the key — so "explicit disable wins" kept a bundled backend off
    # forever (pi314's #40190 symptom). Keys keep every surface aligned.
    plugin_keys = [entry[5] for entry in entries]
    plugin_labels = [
        (f"{name} \u2014 {description}" if description else name)
        + (" [bundled]" if source == "bundled" else "")
        for name, _version, description, source, _d, _key in entries
    ]
    # Selected (enabled) when in enabled-set AND not in disabled-set. Accept the
    # legacy bare name on either side for back-compat with configs written
    # before this normalization.
    plugin_selected = {
        i for i, (name, _v, _desc, _src, _d, key) in enumerate(entries)
        if (key in enabled_set or name in enabled_set)
        and key not in disabled_set
        and name not in disabled_set
    }

    categories = _provider_categories()

    # Non-TTY fallback
    if not sys.stdin.isatty():
        console.print("[dim]Interactive mode requires a terminal.[/dim]")
        return

    # Launch the composite curses UI
    try:
        import curses
        _run_composite_ui(curses, plugin_keys, plugin_labels, plugin_selected,
                          disabled_set, categories, console)
    except ImportError:
        _run_composite_fallback(plugin_keys, plugin_labels, plugin_selected,
                                disabled_set, categories, console)


def _persist_plugin_selection(plugin_keys, chosen, disabled) -> tuple[bool, set]:
    """Save the composite UI's checkbox state; returns ``(changed, new_enabled)``.

    Persist by canonical key. Unchecked plugins are written to the disabled-list so they stay off
    even if a future plugin auto-enables itself — but we ONLY ever write the canonical key (never
    the bare manifest name), so the disabled-list can't drift out of sync with what ``cmd_enable``
    clears or what PluginManager gates on (#40190). Re-checking a plugin also drops any stale
    legacy bare-leaf disable so it is fully cleared from the disabled-list.
    """
    new_enabled: set = set()
    new_disabled: set = set(disabled)  # preserve existing disabled state for unseen plugins
    for i, key in enumerate(plugin_keys):
        if i in chosen:
            new_enabled.add(key)
            _discard_key_and_leaf(new_disabled, key)
        else:
            new_disabled.add(key)

    changed = new_enabled != _get_enabled_set() or new_disabled != disabled
    if changed:
        _save_enabled_set(new_enabled)
        _save_disabled_set(new_disabled)
    return changed, new_enabled


def _run_composite_ui(curses, plugin_keys, plugin_labels, plugin_selected,
                      disabled, categories, console):
    """Custom curses screen with checkboxes + category action rows."""
    from hermes_cli.curses_ui import flush_stdin

    chosen = set(plugin_selected)
    n_plugins = len(plugin_keys)
    # Total rows: plugins + separator + categories
    # separator is not navigable
    n_categories = len(categories)
    total_items = n_plugins + n_categories  # navigable items

    providers_changed = False

    def _init_colors():
        if curses.has_colors():
            curses.start_color()
            curses.use_default_colors()
            curses.init_pair(1, curses.COLOR_GREEN, -1)
            curses.init_pair(2, curses.COLOR_YELLOW, -1)
            curses.init_pair(3, curses.COLOR_CYAN, -1)
            curses.init_pair(4, 8 if curses.COLORS > 8 else curses.COLOR_WHITE, -1)  # dim gray

    def _attr(base, pair):
        attr = base
        if curses.has_colors():
            attr |= curses.color_pair(pair)
        return attr

    def _put(stdscr, y, x, text, max_x, attr):
        try:
            stdscr.addnstr(y, x, text, max_x - 1, attr)
        except curses.error:
            pass

    def _configure_category(ci):
        """Leave curses, run the category's picker, refresh its row, re-enter curses."""
        curses.endwin()
        nonlocal providers_changed
        cat_name, _cat_cur, cat_fn = categories[ci]
        if cat_fn():
            providers_changed = True
            categories[ci] = (cat_name, _provider_categories()[ci][1], cat_fn)
        stdscr = curses.initscr()
        curses.noecho()
        curses.cbreak()
        stdscr.keypad(True)
        _init_colors()
        curses.curs_set(0)
        return stdscr

    def _draw(stdscr):
        curses.curs_set(0)
        _init_colors()
        # key -> new cursor, given (cursor, page_size)
        nav = {}
        for keys, move in (
            ((curses.KEY_UP, ord("k")), lambda c, p: (c - 1) % total_items),
            ((curses.KEY_DOWN, ord("j")), lambda c, p: (c + 1) % total_items),
            ((curses.KEY_NPAGE, ord("f")), lambda c, p: min(total_items - 1, c + p)),
            ((curses.KEY_PPAGE, ord("b")), lambda c, p: max(0, c - p)),
            ((curses.KEY_HOME,), lambda c, p: 0),
            ((curses.KEY_END,), lambda c, p: total_items - 1),
        ):
            nav.update(dict.fromkeys(keys, move))
        cursor = 0
        scroll_offset = 0

        while True:
            stdscr.clear()
            max_y, max_x = stdscr.getmaxyx()

            # Header
            _put(stdscr, 0, 0, "Plugins", max_x, _attr(curses.A_BOLD, 2))
            _put(
                stdscr, 1, 0,
                "  ↑↓/j/k navigate  PgUp/PgDn page  SPACE toggle  ENTER configure/confirm  ESC done",
                max_x, curses.A_DIM,
            )

            # Navigable indices: plugins 0..n_plugins-1, categories n_plugins..total_items-1;
            # section headers / separator are drawn but skipped in scroll math.
            visible_rows = max_y - 4
            if cursor < scroll_offset:
                scroll_offset = cursor
            elif cursor >= scroll_offset + visible_rows:
                scroll_offset = cursor - visible_rows + 1

            # Body rows as (text, attr); "" is a blank separator. Drawn from y=3
            # and truncated at the last screen line.
            lines = []
            if n_plugins > 0:
                lines.append(("  General Plugins", _attr(curses.A_BOLD, 2)))
                for i in range(scroll_offset, min(n_plugins, scroll_offset + max(visible_rows, 0))):
                    check = "\u2713" if i in chosen else " "
                    arrow = "\u2192" if i == cursor else " "
                    attr = _attr(curses.A_BOLD, 1) if i == cursor else curses.A_NORMAL
                    lines.append((f" {arrow} [{check}] {plugin_labels[i]}", attr))
            lines.append(("", curses.A_NORMAL))
            if n_categories > 0:
                lines.append(("  Provider Plugins", _attr(curses.A_BOLD, 2)))
                for ci, (cat_name, cat_current, _cat_fn) in enumerate(categories):
                    cat_idx = n_plugins + ci
                    arrow = "\u2192" if cat_idx == cursor else " "
                    attr = _attr(curses.A_BOLD, 3) if cat_idx == cursor else curses.A_NORMAL
                    lines.append((f" {arrow}   {cat_name:<24} \u25b8 {cat_current}", attr))
            for y, (text, attr) in enumerate(lines[: max(0, max_y - 4)], start=3):
                if text:
                    _put(stdscr, y, 0, text, max_x, attr)

            stdscr.refresh()
            key = stdscr.getch()

            if key in nav:
                if total_items > 0:  # (with no rows, every motion leaves cursor at 0)
                    cursor = nav[key](cursor, max(1, max_y - 5))
            elif key == ord(" ") or key in {curses.KEY_ENTER, 10, 13}:
                if cursor >= n_plugins:
                    # Provider category — launch sub-screen (SPACE and ENTER alike)
                    if cursor - n_plugins < n_categories:
                        stdscr = _configure_category(cursor - n_plugins)
                elif key == ord(" "):
                    chosen.symmetric_difference_update({cursor})
                else:
                    return  # ENTER on a plugin checkbox — confirm and exit
            elif key in {27, ord("q")}:
                return  # plugin changes are saved on exit

    curses.wrapper(_draw)
    flush_stdin()

    changed, new_enabled = _persist_plugin_selection(plugin_keys, chosen, disabled)
    if changed:
        console.print(
            f"\n[green]\u2713[/green] General plugins: {len(new_enabled)} enabled, "
            f"{len(plugin_keys) - len(new_enabled)} disabled."
        )
    elif n_plugins > 0:
        console.print("\n[dim]General plugins unchanged.[/dim]")

    if providers_changed:
        new_memory = _get_current_memory_provider() or "built-in"
        new_context = _get_current_context_engine()
        console.print(
            f"[green]\u2713[/green] Memory provider: [bold]{new_memory}[/bold]  "
            f"Context engine: [bold]{new_context}[/bold]"
        )

    if n_plugins > 0 or providers_changed:
        console.print("[dim]Changes take effect on next session.[/dim]")
    console.print()


def _run_composite_fallback(plugin_keys, plugin_labels, plugin_selected,
                            disabled, categories, console):
    """Text-based fallback for the composite plugins UI."""
    from hermes_cli.colors import Colors, color

    print(color("\n  Plugins", Colors.YELLOW))

    # General plugins
    if plugin_keys:
        chosen = set(plugin_selected)
        print(color("\n  General Plugins", Colors.YELLOW))
        print(color("  Toggle by number, Enter to confirm.\n", Colors.DIM))

        while True:
            for i, label in enumerate(plugin_labels):
                marker = color("[\u2713]", Colors.GREEN) if i in chosen else "[ ]"
                print(f"  {marker} {i + 1:>2}. {label}")
            print()
            try:
                val = input(color("  Toggle # (or Enter to confirm): ", Colors.DIM)).strip()
                if not val:
                    break
                idx = int(val) - 1
                if 0 <= idx < len(plugin_keys):
                    chosen.symmetric_difference_update({idx})
            except (ValueError, KeyboardInterrupt, EOFError):
                return
            print()

        _persist_plugin_selection(plugin_keys, chosen, disabled)

    # Provider categories
    if categories:
        print(color("\n  Provider Plugins", Colors.YELLOW))
        for ci, (cat_name, cat_current, cat_fn) in enumerate(categories):
            print(f"  {ci + 1}. {cat_name} [{cat_current}]")
        print()
        try:
            val = input(color("  Configure # (or Enter to skip): ", Colors.DIM)).strip()
            if val:
                ci = int(val) - 1
                if 0 <= ci < len(categories):
                    categories[ci][2]()  # call the configure function
        except (ValueError, KeyboardInterrupt, EOFError):
            pass

    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."""
    warnings: list[str] = []
    try:
        insecure = _resolve_git_url(identifier)[0].startswith(("http://", "file://"))
    except ValueError:
        insecure = False
    if insecure:
        warnings.append("Insecure URL scheme; prefer https:// or git@ for production installs.")

    try:
        target, installed_manifest, installed_name = _install_plugin_core(
            identifier,
            force=force,
        )
    except PluginScanBlocked as exc:
        fields = ("pattern_id", "severity", "category", "file", "line", "description")
        findings = [
            {k: getattr(f, k) for k in fields}
            for f in (exc.scan_result.findings if exc.scan_result is not None else ())
        ]
        return {
            "ok": False,
            "error": str(exc),
            "scan_blocked": True,
            "scan_verdict": getattr(exc.scan_result, "verdict", "dangerous"),
            "scan_findings": findings,
        }
    except PluginOperationError as exc:
        return {"ok": False, "error": str(exc)}

    missing_env = _missing_requires_env_names(installed_manifest)
    if enable:
        _set_plugin_enabled(installed_name, enable=True)

    ap = target / "after-install.md"
    hint = str(ap) if ap.exists() else None

    return {
        "ok": True,
        "plugin_name": installed_name,
        "warnings": warnings,
        "missing_env": missing_env,
        "after_install_path": hint,
        "enabled": enable,
    }


def _get_plugin_toolset_key(name: str) -> Optional[str]:
    """Return the toolset key a plugin registers its tools under, or None.

    Queries the live tool registry — the plugin must already be loaded. Falls back to reading
    ``provides_tools`` from plugin.yaml and looking up the toolset from the registry for the first
    tool name found.
    """
    try:
        from tools.registry import registry
    except Exception:
        return None

    def _first_toolset(tool_names) -> Optional[str]:
        for tool_name in tool_names:
            entry = registry.get_entry(tool_name)
            if entry and entry.toolset:
                return entry.toolset
        return None

    def _from_loaded_plugin() -> Optional[str]:
        from hermes_cli.plugins import discover_plugins, get_plugin_manager
        discover_plugins()  # idempotent — ensures plugins are loaded
        for _key, loaded in get_plugin_manager()._plugins.items():
            if loaded.manifest.name == name or _key == name:
                return _first_toolset(loaded.tools_registered)
        return None

    def _from_manifest_on_disk() -> Optional[str]:
        from hermes_cli.plugins import get_bundled_plugins_dir
        for base in (get_bundled_plugins_dir(), _plugins_dir()):
            candidate = base / name
            if base.is_dir() and candidate.is_dir():
                toolset = _first_toolset(_read_manifest(candidate).get("provides_tools") or [])
                if toolset:
                    return toolset
        return None

    for lookup in (_from_loaded_plugin, _from_manifest_on_disk):
        try:
            toolset = lookup()
        except Exception:
            continue
        if toolset:
            return toolset
    return None


def _toggle_plugin_toolset(name: str, *, enable: bool) -> None:
    """Add or remove a plugin's toolset from platform_toolsets for all platforms.

    Only acts if the plugin actually provides tools (has a toolset key).
    """
    toolset_key = _get_plugin_toolset_key(name)
    if not toolset_key:
        return

    from hermes_cli.config import load_config, save_config

    config = load_config()
    platform_toolsets = _sub_dict(config, "platform_toolsets")

    changed = False
    for platform, ts_list in platform_toolsets.items():
        if not isinstance(ts_list, list):
            continue
        if enable:
            if toolset_key not in ts_list:
                ts_list.append(toolset_key)
                changed = True
        elif toolset_key in ts_list:
            ts_list.remove(toolset_key)
            changed = True

    # If enabling and no platforms have toolset lists yet, add to "cli" at minimum
    if enable and not changed and not platform_toolsets:
        platform_toolsets["cli"] = [toolset_key]
        changed = True

    if changed:
        save_config(config)


def dashboard_set_agent_plugin_enabled(name: str, *, enabled: bool) -> dict[str, Any]:
    """Enable or disable a plugin in ``config.yaml`` (runtime allow/deny lists)."""
    if _resolve_plugin_key(name) is None:
        return {"ok": False, "error": f"Plugin '{name}' is not installed or bundled."}

    en = _get_enabled_set()
    dis = _get_disabled_set()
    already = (name in en and name not in dis) if enabled else (name not in en and name in dis)
    if already:
        return {"ok": True, "name": name, "unchanged": True}
    _set_plugin_enabled(name, enable=enabled)
    _toggle_plugin_toolset(name, enable=enabled)
    return {"ok": True, "name": name, "unchanged": False}


def _user_installed_plugin_dir(name: str) -> Optional[Path]:
    """Resolved path under ``~/.hermes/plugins/<name>`` if it exists."""
    plugins_dir = _plugins_dir()
    try:
        target = _sanitize_plugin_name(name, plugins_dir, allow_subdir=True)
    except ValueError:
        return None
    return target if target.is_dir() else None


def dashboard_update_user_plugin(name: str) -> dict[str, Any]:
    """``git pull`` inside ``~/.hermes/plugins/<name>``."""
    target = _user_installed_plugin_dir(name)
    if target is None:
        return {
            "ok": False,
            "error": f"Plugin '{name}' was not found under {_plugins_dir()}.",
        }

    try:
        msg = _pull_plugin_update(
            target,
            lambda rec: (
                f"Plugin '{name}' is pinned to {rec.get('revision')}; "
                f"run `hermes plugins install {rec.get('source', '<source>')} --force "
                "--ref <40-character commit SHA>` to move it."
            ),
            lambda: f"Plugin '{name}' is not a git checkout; cannot pull updates.",
        )
    except PluginOperationError as exc:
        return {"ok": False, "error": str(exc)}

    _post_pull_housekeeping(target, _console())
    unchanged = "Already up to date" in msg
    return {"ok": True, "name": name, "output": msg, "unchanged": unchanged}


def _clear_plugin_bytecode(target: Path) -> int:
    """Remove ``__pycache__`` dirs under a just-updated plugin checkout.

    Plugin dirs live outside the main repo, so the launch-time checkout fingerprint sweep in
    ``hermes_cli.main`` never covers them. After a ``git pull`` changes a plugin's ``.py`` files,
    stale bytecode here can produce the same ImportError class as #6207/#60242 in whichever process
    imports the plugin next. Never raises.
    """
    removed = 0
    try:
        for cache_dir in target.rglob("__pycache__"):
            if cache_dir.is_dir():
                try:
                    shutil.rmtree(cache_dir)
                    removed += 1
                except OSError:
                    pass
    except OSError:
        pass
    return removed


def _run_plugin_git(
    git_exe: str, target: Path, *args: str, timeout: int = 60
) -> subprocess.CompletedProcess:
    """Run one git command inside a plugin checkout (non-interactive)."""
    return subprocess.run(
        [git_exe, *args],
        capture_output=True,
        text=True, encoding='utf-8', errors='replace',
        timeout=timeout,
        cwd=str(target),
        stdin=subprocess.DEVNULL,
        env=noninteractive_git_env(),
    )


def _stash_ref(git_exe: str, target: Path) -> str:
    """Current ``refs/stash`` commit, or empty string when no stash exists."""
    probe = _run_plugin_git(git_exe, target, "rev-parse", "--verify", "refs/stash")
    return probe.stdout.strip() if probe.returncode == 0 else ""


def _reapply_stash(git_exe: str, target: Path) -> bool:
    """``stash apply`` the autostash; drop it on a clean apply. False when it applied with
    errors or left unmerged paths (the stash entry is kept in that case)."""
    restore = _run_plugin_git(git_exe, target, "stash", "apply", "stash@{0}")
    unmerged = _run_plugin_git(git_exe, target, "diff", "--name-only", "--diff-filter=U")
    if restore.returncode != 0 or unmerged.stdout.strip():
        return False
    _run_plugin_git(git_exe, target, "stash", "drop", "stash@{0}")
    return True


def _git_pull_plugin_dir(target: Path) -> tuple[bool, str]:
    """``git pull --ff-only`` a plugin checkout, autostashing local edits.

    Users tweak installed plugins in place (config constants, small patches), and a plain ``pull
    --ff-only`` then aborts with "Your local changes ... would be overwritten by merge" — making the
    plugin permanently un-updatable until they hand-run git.
    """
    git_exe = _resolve_git_executable()
    if not git_exe:
        return False, "git is not installed or not in PATH."
    try:
        status = _run_plugin_git(git_exe, target, "status", "--porcelain")
        dirty = status.returncode == 0 and bool(status.stdout.strip())

        stash_created = False
        pre_stash = ""
        if dirty:
            pre_stash = _stash_ref(git_exe, target)
            push = _run_plugin_git(
                git_exe, target,
                "stash", "push", "--include-untracked",
                "-m", "hermes-plugin-update-autostash",
            )
            post_stash = _stash_ref(git_exe, target)
            stash_created = bool(post_stash) and post_stash != pre_stash
            if not stash_created:
                # Nothing was saved — do not risk the pull clobbering edits.
                err = _safe_git_error(push)
                return False, (
                    "Local changes in the plugin checkout could not be "
                    "stashed; update aborted before touching the checkout."
                    + (f"\n{err}" if err else "")
                )
            if push.returncode != 0:
                # Saved-but-couldn't-clean (undeletable untracked files):
                # the stash entry is complete; reset tracked mods so the
                # pull isn't blocked by a still-dirty tree.
                _run_plugin_git(git_exe, target, "reset", "--hard", "HEAD")

        result = _run_plugin_git(git_exe, target, "pull", "--ff-only")

        if result.returncode != 0:
            err = _safe_git_error(result) or "git pull failed."
            if stash_created:
                # Put the user's edits back before reporting the failure.
                if _reapply_stash(git_exe, target):
                    note = "Local changes were restored."
                else:
                    note = (
                        "Local changes are preserved in git stash "
                        "(restore with: git stash pop)."
                    )
                return False, f"{err}\n{note}"
            return False, err

        pulled = result.stdout.strip()
        if not stash_created:
            return True, pulled
        if _reapply_stash(git_exe, target):
            return True, pulled + "\nLocal changes were re-applied on top of the update."

        # Conflicted re-apply: leave the plugin importable on the updated
        # revision; the user's edits stay safe in the stash entry.
        _run_plugin_git(git_exe, target, "reset", "--hard", "HEAD")
        return True, pulled + (
            "\n⚠ Local changes in this plugin conflicted with the update and "
            "were NOT re-applied. They are preserved in git stash — inspect "
            "with `git stash show -p stash@{0}` and re-apply with "
            f"`git stash pop` inside {target}."
        )
    except FileNotFoundError:
        return False, "git is not installed or not in PATH."
    except subprocess.TimeoutExpired:
        return False, "Git operation timed out after 60 seconds."


def dashboard_remove_user_plugin(name: str) -> dict[str, Any]:
    """Delete a plugin tree under ``~/.hermes/plugins/`` only."""
    plugins_dir = _plugins_dir()
    for n, _ver, _d, src, _path, _key in _discover_all_plugins():
        if n == name and src == "bundled":
            return {"ok": False, "error": "Bundled plugins cannot be removed from the dashboard."}

    target = _user_installed_plugin_dir(name)
    if target is None:
        return {
            "ok": False,
            "error": f"Plugin '{name}' was not found under {plugins_dir}.",
        }

    try:
        _remove_plugin_core(target)
    except (OSError, PluginOperationError) as exc:
        return {"ok": False, "error": f"Could not remove plugin '{name}': {exc}"}
    return {"ok": True, "name": name}


def cmd_plugin_doctor(target: str = ".", *, ci: bool = False) -> None:
    """Validate one plugin through runtime discovery and registration."""
    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 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] "
            f"[dim](index source: {source})[/dim]"
        )
        return

    from rich.table import Table

    table = Table(title=f"Community plugins ({len(results)} match{'es' if len(results) != 1 else ''})")
    table.add_column("Name", style="bold")
    table.add_column("Description")
    table.add_column("Author")
    table.add_column("Tags", style="dim")
    for e in results:
        desc = e.description
        if len(desc) > 70:
            desc = desc[: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)."""
    if getattr(args, yes_attr, False):
        return True
    if getattr(args, no_attr, False):
        return False
    return None


def _action_pack(args):
    from hermes_cli.plugin_packs import pack_command

    pack_command(args)


# Tri-state flags: neither --x nor --no-x given == None == interactive prompt.
_PLUGIN_ACTIONS = {
    "install": lambda args: cmd_install(
        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),
    ),
    "update": lambda args: cmd_update(args.name),
    "remove": lambda args: cmd_remove(args.name),
    "rm": lambda args: cmd_remove(args.name),
    "uninstall": lambda args: cmd_remove(args.name),
    "enable": lambda args: cmd_enable(
        args.name,
        allow_tool_override=_tri_state_flag(args, "allow_tool_override", "no_allow_tool_override"),
    ),
    "disable": lambda args: cmd_disable(args.name),
    "capabilities": lambda args: cmd_capabilities(getattr(args, "name", None)),
    "list": lambda args: cmd_list(args),
    "ls": lambda args: cmd_list(args),
    "doctor": lambda args: cmd_plugin_doctor(args.target, ci=getattr(args, "ci", False)),
    "pack": _action_pack,
    "show": lambda args: cmd_show(args.name),
    "info": lambda args: cmd_show(args.name),
    None: lambda args: cmd_toggle(),
}


def plugins_command(args) -> None:
    """Dispatch hermes plugins subcommands."""
    action = getattr(args, "plugins_action", None)
    handler = _PLUGIN_ACTIONS.get(action)
    if handler is None:
        _console().print(f"[red]Unknown plugins action: {action}[/red]")
        sys.exit(1)
    handler(args)