hermes-agent/hermes_cli/plugins_cmd_update_all.py

"""``hermes plugins update`` with no name: update every installed plugin that has an update.

The list comes from the read-only ``check-updates`` probe (:func:`hermes_cli.plugins_updates.run_checks`);
each plugin is then updated through the single-plugin
path (:func:`~hermes_cli.plugins_cmd_update.update_one`), so catalog consent, the security scan and
capability re-consent still apply per plugin.

Sibling of :mod:`hermes_cli.plugins_cmd` (the facade is imported late, never at module level).
"""

from __future__ import annotations

import logging
import sys
from dataclasses import dataclass, field
from typing import Any

# Only these classes have an update path `hermes plugins update <name>` can drive: a catalog install
# re-pins to the reviewed sha, a tracked git install pulls. A pip row is a PyPI distribution that
# Hermes never upgrades (#132494), so offering it would promise an update nothing applies.
_UPDATE_KIND = {"catalog": "catalog re-pin", "git": "git pull"}
logger = logging.getLogger(__name__)


def _pc():
    from hermes_cli import plugins_cmd
    return plugins_cmd


@dataclass
class UpdatePlan:
    """What ``update`` with no name would do: *movable* rows it updates, *skipped* ``(row, reason)``
    pairs it cannot move, *current* the count already up to date."""

    movable: list[Any] = field(default_factory=list)
    skipped: list[tuple[Any, str]] = field(default_factory=list)
    current: int = 0


def plan_update_all(results: list[Any], installs: dict | None = None) -> UpdatePlan:
    """Split ``run_checks`` rows into the plugins ``update`` can move and the ones it cannot (with why).
    *installs* (the install records the check ran against) lets a catalog install held with ``--ref``
    be skipped: only naming it (``hermes plugins update <name>``) moves it."""
    from hermes_cli.plugins_auto_update import HELD_WITH_REF, held_with_ref
    from hermes_cli.plugins_updates import held_for_newer_hermes
    plan = UpdatePlan()
    installs = installs or {}
    for row in results:
        if row.needs_fixing:
            plan.skipped.append((row, f"needs fixing: {row.needs_fixing}"))
        elif row.update_available is True and (held := held_for_newer_hermes(row)):
            plan.skipped.append((row, f"update available after updating Hermes ({held})"))
        elif row.update_available is True and held_with_ref(installs.get(row.name)):
            plan.skipped.append((row, HELD_WITH_REF.format(name=row.name)))
        elif row.update_available is True and row.klass in _UPDATE_KIND:
            plan.movable.append(row)
        elif row.update_available is True:
            plan.skipped.append((row, f"{row.klass} distribution; Hermes does not upgrade it"))
        elif row.update_available is None:
            plan.skipped.append((row, row.reason or "update status unknown"))
        elif row.reason:  # up to date by construction but worth naming (pinned)
            plan.skipped.append((row, row.reason))
        else:
            plan.current += 1
    return plan


def _short(reason: str) -> str:
    first = reason.strip().splitlines()[0] if reason.strip() else ""
    return first if len(first) <= 200 else first[:197] + "..."


def _update_one(row, console, *, interactive: bool) -> tuple[str, str]:
    """Run the single-plugin update for one plan row; return ``(status, detail)``, status being
    updated/current/failed. A refusal is printed like ``update <name>`` prints it, then recorded."""
    from rich.markup import escape
    from hermes_cli.plugins_cmd_catalog import RepinConsentRequired
    from hermes_cli.plugins_cmd_update import update_one
    target = _pc()._user_installed_plugin_dir(row.name)
    if target is None:
        return "failed", "no longer installed"
    try:
        moved = update_one(row.name, target, console, interactive=interactive)
    except RepinConsentRequired as exc:
        console.print(f"[yellow]Update of {escape(row.name)} not applied:[/yellow] {escape(str(exc))}")
        return "failed", _short(f"not applied: {exc}")
    except _pc().PluginOperationError as exc:
        console.print(f"[red]Error:[/red] {escape(str(exc))}")
        return "failed", _short(str(exc))
    except Exception as exc:  # per-plugin batch boundary: a feed timeout or crash must not stop the rest
        logger.warning("update of plugin %s failed", row.name, exc_info=True)
        console.print(f"[red]Error:[/red] {escape(type(exc).__name__)}: {escape(str(exc))}")
        return "failed", _short(f"{type(exc).__name__}: {exc}")
    if not moved:
        return "current", ""
    return "updated", f"{(row.current or '')[:8]}..{(row.latest or '')[:8]}"


def _print_plan(console, plan: UpdatePlan) -> None:
    from rich.markup import escape
    table = _pc()._table([("#", "dim"), ("Plugin", "bold"), ("Current", ""), ("Latest", ""), ("Update", "")],
                         title="Plugin updates available")
    for i, row in enumerate(plan.movable, 1):
        table.add_row(str(i), row.name, (row.current or "-")[:12], (row.latest or "-")[:12], _UPDATE_KIND[row.klass])
    console.print()
    console.print(table)
    for row, reason in plan.skipped:
        console.print(f"  [dim]skipped {escape(row.name)} ({row.klass}): {escape(reason)}[/dim]")


def cmd_update_all(*, yes: bool = False) -> None:
    """``hermes plugins update [--yes]``: list the plugins with an update and update them all.

    A terminal asks once (``Update all N? [Y/n]``); without one nothing moves unless ``--yes`` is
    given. Each plugin then goes through ``update_one``, interactive only when a terminal is present,
    so per-plugin consent prompts still appear there and stay closed (fail closed) under ``--yes``
    in scripts. One failure does not stop the rest; exit 1 when any update failed.
    """
    from rich.markup import escape
    from hermes_cli.plugins_updates import run_checks

    console = _pc()._console()
    console.print("[dim]Checking installed plugins for updates...[/dim]")
    from hermes_cli.plugins_cadence import snapshot_installs
    plugins_dir = _pc()._plugins_dir()
    checked = snapshot_installs(plugins_dir)
    plan = plan_update_all(run_checks(plugins_dir), checked)
    if not plan.movable:
        for row, reason in plan.skipped:
            console.print(f"  [dim]skipped {escape(row.name)} ({row.klass}): {escape(reason)}[/dim]")
        console.print("[green]✓[/green] No plugin updates to apply.")
        return
    _print_plan(console, plan)
    count = len(plan.movable)
    tty = _pc()._is_tty()
    if not yes:
        if not tty:
            _pc()._fail(console, f"[yellow]Not updating: no terminal to confirm. Re-run with --yes to update "
                                 f"these {count} plugin(s).[/yellow]")
        try:
            answer = input(f"\nUpdate all {count}? [Y/n]: ").strip().lower()
        except (EOFError, KeyboardInterrupt):
            answer = "n"
        if answer not in ("", "y", "yes"):
            console.print("Cancelled. Nothing was updated.")
            return

    outcomes = []
    for row in plan.movable:
        console.print(f"\n[bold]── {row.name}[/bold]")
        outcomes.append((row.name, *_update_one(row, console, interactive=tty)))

    console.print("\n[bold]Summary[/bold]")
    marks = {"updated": "[green]✓ updated[/green]", "current": "[dim]= already current[/dim]",
             "failed": "[red]✗ failed[/red]"}
    for name, status, detail in outcomes:
        console.print(f"  {marks[status]}  {escape(name)}" + (f"  [dim]{escape(detail)}[/dim]" if detail else ""),
                      highlight=False)
    if any(status == "failed" for _, status, _ in outcomes):
        sys.exit(1)