779 lines
38 KiB
Python
779 lines
38 KiB
Python
"""Welcome banner, ASCII art, skills summary, and update check for the CLI."""
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
from hermes_cli import source_check
|
|
# Historical updater import (tests/compat/old_updater_surface.json). In-tree callers use the owner.
|
|
from hermes_cli.source_check import _github_compare_behind # noqa: F401
|
|
from hermes_constants import get_hermes_home
|
|
from typing import TYPE_CHECKING, Any, Dict, List, Optional
|
|
|
|
# rich and prompt_toolkit are imported lazily: this module sits on the TUI gateway's critical
|
|
# startup path purely for the lightweight update-check helpers, and eager rich/prompt_toolkit
|
|
# imports cost ~50ms before ``gateway.ready`` could fire.
|
|
if TYPE_CHECKING:
|
|
from rich.console import Console
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# ANSI building blocks for conversation display (``_DIM``/``_RST`` are imported by callbacks.py).
|
|
_DIM = "\033[2m"
|
|
_RST = "\033[0m"
|
|
|
|
|
|
def _check_via_pypi() -> Optional[int]:
|
|
# Shim to stop the old updater doing work until relaunch. no registry query.
|
|
return None
|
|
|
|
|
|
def check_via_pypi() -> Optional[int]:
|
|
# Shim to stop the old updater doing work until relaunch. status is unknown.
|
|
return None
|
|
|
|
|
|
def _quiet(fn, default=None):
|
|
"""``fn()``, or ``default`` on any exception — for best-effort display inputs."""
|
|
try:
|
|
return fn()
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def cprint(text: str):
|
|
"""Print ANSI-colored text through prompt_toolkit's renderer."""
|
|
from prompt_toolkit import print_formatted_text as _pt_print
|
|
from prompt_toolkit.formatted_text import ANSI as _PT_ANSI
|
|
# prompt_toolkit needs a real console: on Windows a redirected/absent stdout raises
|
|
# NoConsoleScreenBufferError, and display helpers must never crash the caller over that.
|
|
if _quiet(lambda: _pt_print(_PT_ANSI(text)) or True) is None:
|
|
print(text)
|
|
|
|
|
|
def _active_skin():
|
|
"""The active skin object (raises when the skin engine is unavailable)."""
|
|
from hermes_cli.skin_engine import get_active_skin
|
|
return get_active_skin()
|
|
|
|
|
|
def _skin_color(key: str, fallback: str) -> str:
|
|
"""Get a color from the active skin, or return fallback."""
|
|
return _quiet(lambda: _active_skin().get_color(key, fallback), fallback)
|
|
|
|
|
|
# === ASCII Art & Branding ===
|
|
|
|
from hermes_cli import __release_date__ as RELEASE_DATE
|
|
from hermes_cli.version_info import get_version_info
|
|
|
|
HERMES_AGENT_LOGO = """[bold #FFD700]██╗ ██╗███████╗██████╗ ███╗ ███╗███████╗███████╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/]
|
|
[bold #FFD700]██║ ██║██╔════╝██╔══██╗████╗ ████║██╔════╝██╔════╝ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/]
|
|
[#FFBF00]███████║█████╗ ██████╔╝██╔████╔██║█████╗ ███████╗█████╗███████║██║ ███╗█████╗ ██╔██╗ ██║ ██║[/]
|
|
[#FFBF00]██╔══██║██╔══╝ ██╔══██╗██║╚██╔╝██║██╔══╝ ╚════██║╚════╝██╔══██║██║ ██║██╔══╝ ██║╚██╗██║ ██║[/]
|
|
[#CD7F32]██║ ██║███████╗██║ ██║██║ ╚═╝ ██║███████╗███████║ ██║ ██║╚██████╔╝███████╗██║ ╚████║ ██║[/]
|
|
[#CD7F32]╚═╝ ╚═╝╚══════╝╚═╝ ╚═╝╚═╝ ╚═╝╚══════╝╚══════╝ ╚═╝ ╚═╝ ╚═════╝ ╚══════╝╚═╝ ╚═══╝ ╚═╝[/]"""
|
|
|
|
HERMES_CADUCEUS = """[#CD7F32]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⡀⠀⣀⣀⠀⢀⣀⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#CD7F32]⠀⠀⠀⠀⠀⠀⢀⣠⣴⣾⣿⣿⣇⠸⣿⣿⠇⣸⣿⣿⣷⣦⣄⡀⠀⠀⠀⠀⠀⠀[/]
|
|
[#FFBF00]⠀⢀⣠⣴⣶⠿⠋⣩⡿⣿⡿⠻⣿⡇⢠⡄⢸⣿⠟⢿⣿⢿⣍⠙⠿⣶⣦⣄⡀⠀[/]
|
|
[#FFBF00]⠀⠀⠉⠉⠁⠶⠟⠋⠀⠉⠀⢀⣈⣁⡈⢁⣈⣁⡀⠀⠉⠀⠙⠻⠶⠈⠉⠉⠀⠀[/]
|
|
[#FFD700]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣴⣿⡿⠛⢁⡈⠛⢿⣿⣦⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#FFD700]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠿⣿⣦⣤⣈⠁⢠⣴⣿⠿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#FFBF00]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠉⠻⢿⣿⣦⡉⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#FFBF00]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠘⢷⣦⣈⠛⠃⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#CD7F32]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢠⣴⠦⠈⠙⠿⣦⡄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#CD7F32]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠸⣿⣤⡈⠁⢤⣿⠇⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠉⠛⠷⠄⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⠑⢶⣄⡀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⣿⠁⢰⡆⠈⡿⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠳⠈⣡⠞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
|
|
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]"""
|
|
|
|
# === Skills scanning ===
|
|
|
|
# Per-process caches: ``None`` until computed, then a 1-tuple ``(value,)`` so a computed ``None``
|
|
# is distinguishable from "not yet computed". Reset by assigning ``None`` (tests, ``hermes skills``).
|
|
_available_skills_cache: Optional[tuple] = None
|
|
_git_banner_state_cache: Optional[tuple] = None
|
|
_latest_release_cache: Optional[tuple] = None
|
|
|
|
_UNCACHED = object() # compute() result that must not be memoized
|
|
|
|
|
|
def _memo(cache_name: str, compute):
|
|
"""Return the cached value under module global ``cache_name``, computing (and storing) it once.
|
|
|
|
Not consulted under a routed profile (HERMES_HOME override): every memo here is derived from the
|
|
launch home (its skills tree, its checkout), and the TUI gateway calls these per profile."""
|
|
from hermes_constants import get_hermes_home_override
|
|
if get_hermes_home_override() is not None:
|
|
return compute()
|
|
cached = globals()[cache_name]
|
|
if cached is not None:
|
|
return cached[0]
|
|
value = compute()
|
|
if value is not _UNCACHED:
|
|
globals()[cache_name] = (value,)
|
|
return value
|
|
|
|
|
|
def get_available_skills() -> Dict[str, List[str]]:
|
|
"""Return skills grouped by category, filtered by platform and disabled state.
|
|
|
|
Cached per-process (the skills-tree walk costs ~100ms and feeds only the startup banner);
|
|
``prefetch_banner_data()`` pays it off-thread. A failed scan yields ``{}`` and is not cached.
|
|
"""
|
|
def _scan():
|
|
from tools.skills_tool import _find_all_skills
|
|
return _find_all_skills() # already filtered
|
|
|
|
def _compute():
|
|
all_skills = _quiet(_scan)
|
|
if all_skills is None:
|
|
return _UNCACHED
|
|
skills_by_category: Dict[str, List[str]] = {}
|
|
for skill in all_skills:
|
|
skills_by_category.setdefault(skill.get("category") or "general", []).append(skill["name"])
|
|
return skills_by_category
|
|
result = _memo("_available_skills_cache", _compute)
|
|
return {} if result is _UNCACHED else result
|
|
|
|
|
|
def _resolve_repo_dir() -> Optional[Path]:
|
|
"""The active Hermes git checkout, or None if this isn't a git install.
|
|
|
|
Prefers the running code's location: ``$HERMES_HOME/hermes-agent/`` may be a stale copy
|
|
carried over by ``--clone-all``.
|
|
"""
|
|
repo_dir = Path(__file__).parent.parent.resolve()
|
|
if not (repo_dir / ".git").exists():
|
|
repo_dir = get_hermes_home() / "hermes-agent"
|
|
return repo_dir if (repo_dir / ".git").exists() else None
|
|
|
|
|
|
def get_git_banner_state(repo_dir: Optional[Path] = None) -> Optional[dict]:
|
|
"""Return upstream/local git hashes for the startup banner.
|
|
|
|
Cached per-process (default ``repo_dir`` only): 2-3 git subprocesses (~100ms) whose result
|
|
cannot change under a running CLI. The cache lets ``prefetch_banner_data()`` pay it off-thread.
|
|
"""
|
|
if repo_dir is not None:
|
|
return _compute_git_banner_state(repo_dir)
|
|
return _memo("_git_banner_state_cache", _compute_git_banner_state)
|
|
|
|
|
|
def _baked_banner_state() -> Optional[dict]:
|
|
"""Banner state from the baked build SHA (Docker image path), or None."""
|
|
def _baked():
|
|
from hermes_cli.version_info import get_code_identity
|
|
return get_code_identity().get("short_sha")
|
|
baked = _quiet(_baked)
|
|
return {"upstream": baked, "local": baked, "ahead": 0} if baked else None
|
|
|
|
|
|
def _compute_git_banner_state(repo_dir: Optional[Path] = None) -> Optional[dict]:
|
|
repo_dir = repo_dir or _resolve_repo_dir()
|
|
if repo_dir is None:
|
|
return _baked_banner_state()
|
|
upstream, local = (source_check._git_stdout(["rev-parse", "--short=8", rev], cwd=repo_dir) for rev in ("origin/main", "HEAD"))
|
|
if not upstream or not local:
|
|
# Live-git lookup failed (e.g. shallow clone without origin/main).
|
|
return _baked_banner_state()
|
|
ahead = source_check._git_count(["rev-list", "--count", "origin/main..HEAD"], cwd=repo_dir) or 0
|
|
return {"upstream": upstream, "local": local, "ahead": max(ahead, 0)}
|
|
|
|
|
|
_RELEASE_URL_BASE = "https://github.com/NousResearch/hermes-agent/releases/tag"
|
|
|
|
|
|
def get_latest_release_tag(repo_dir: Optional[Path] = None) -> Optional[tuple]:
|
|
"""Return ``(tag, release_url)`` for the latest local git tag, or None (a miss is cached too).
|
|
|
|
Release URL always points at the canonical NousResearch/hermes-agent repo (forks get no link).
|
|
"""
|
|
def _compute():
|
|
rd = repo_dir or _resolve_repo_dir()
|
|
tag = source_check._git_stdout(["describe", "--tags", "--abbrev=0"], cwd=rd, timeout=3) if rd else None
|
|
return (tag, f"{_RELEASE_URL_BASE}/{tag}") if tag else None
|
|
return _memo("_latest_release_cache", _compute)
|
|
|
|
|
|
def format_banner_version_label() -> str:
|
|
"""Return the version label shown in the startup banner title."""
|
|
from hermes_cli.config import get_project_root
|
|
from hermes_cli.steward import read_install_stamp
|
|
from hermes_cli.update_channel import is_canary_tag
|
|
|
|
stamp = read_install_stamp(get_project_root())
|
|
if stamp.get("distribution") == "desktop-app":
|
|
label = f"Hermes Agent v{get_version_info().derived_version}"
|
|
if stamp.get("source") == "commit-build":
|
|
return f"{label} · commit-build · {str(stamp.get('commit') or '')[:12]}"
|
|
if stamp.get("tag"):
|
|
channel = "canary" if is_canary_tag(stamp["tag"]) else "stable"
|
|
return f"{label} · {channel}"
|
|
if stamp.get("payload") == "bootstrap":
|
|
# The old installer shell's CLI backend: the shell never updates
|
|
# itself (`self`), the managed checkout under it does. Name the
|
|
# shell so it doesn't read as a plain packaged build.
|
|
return f"{label} · installer"
|
|
return label
|
|
|
|
base = f"Hermes Agent v{get_version_info().derived_version} ({RELEASE_DATE})"
|
|
from hermes_cli.config import load_config
|
|
from hermes_cli.update_channel import resolve_update_channel
|
|
|
|
channel = resolve_update_channel(_quiet(load_config), get_project_root())
|
|
if channel != "main":
|
|
head = source_check._git_stdout(["rev-parse", "HEAD"], cwd=get_project_root())
|
|
return f"{base} · {channel}" + (f" · local {head[:12]}" if head else "")
|
|
state = get_git_banner_state()
|
|
if not state:
|
|
return base
|
|
upstream, local = state["upstream"], state["local"]
|
|
ahead = int(state.get("ahead") or 0)
|
|
if ahead <= 0 or upstream == local:
|
|
return f"{base} · upstream {upstream}"
|
|
return f"{base} · upstream {upstream} · local {local} (+{ahead} carried {_plural(ahead, 'commit')})"
|
|
|
|
|
|
# === Non-blocking update check ===
|
|
|
|
_update_result: Optional[int] = None
|
|
_update_check_done = threading.Event()
|
|
|
|
|
|
def _daemon(name: Optional[str], target) -> None:
|
|
"""Start a daemon thread running ``target`` with any exception swallowed."""
|
|
threading.Thread(target=lambda: _quiet(target), name=name, daemon=True).start()
|
|
|
|
|
|
def _skip_background_prefetch() -> bool:
|
|
"""True when the banner's background prefetch threads must not start.
|
|
|
|
Under pytest the prefetch daemon threads shell out to git (``rev-parse``,
|
|
``remote get-url``, the banner's git state) at an arbitrary point after
|
|
import, and any test that patches the process-wide ``subprocess`` singleton
|
|
(``patch("subprocess.run")`` / ``patch("subprocess.Popen")``) can record
|
|
that stray spawn in place of the call it meant to pin. Importing
|
|
``tui_gateway.server`` starts this prefetch, which is what flaked
|
|
tests/tui_gateway/test_bot_relay_methods.py.
|
|
Nothing under pytest needs a live update check; tests that exercise the
|
|
prefetch itself monkeypatch this predicate to False.
|
|
|
|
``PYTEST_CURRENT_TEST`` is only set while a test runs, not during
|
|
collection-time imports, hence the ``sys.modules`` check as well.
|
|
"""
|
|
return "PYTEST_CURRENT_TEST" in os.environ or "pytest" in sys.modules
|
|
|
|
|
|
def prefetch_update_check():
|
|
"""Kick off update check in a background daemon thread.
|
|
|
|
No-op under pytest — see ``_skip_background_prefetch``.
|
|
"""
|
|
if _skip_background_prefetch():
|
|
_update_check_done.set()
|
|
return
|
|
|
|
def _run():
|
|
global _update_result
|
|
_update_result = source_check.check_for_updates(passive=True).get("behind")
|
|
_update_check_done.set()
|
|
_daemon(None, _run)
|
|
|
|
|
|
_banner_data_prefetch_started = False
|
|
|
|
|
|
def prefetch_banner_data():
|
|
"""Warm the banner's subprocess/I/O-heavy inputs in a daemon thread.
|
|
|
|
Git state (~130ms) and the skills index (~110ms) are cached per-process by their own modules,
|
|
so warming them while the main thread pays the CPU-bound imports overlaps GIL-releasing I/O
|
|
with import work. Idempotent; failures don't matter because the banner recomputes anything missing.
|
|
"""
|
|
global _banner_data_prefetch_started
|
|
if _banner_data_prefetch_started:
|
|
return
|
|
if _skip_background_prefetch():
|
|
# Same stray-git-spawn cross-talk class as prefetch_update_check:
|
|
# get_git_banner_state() shells out via the shared subprocess
|
|
# singleton from a daemon thread, poisoning process-wide subprocess
|
|
# mocks in unrelated tests.
|
|
_banner_data_prefetch_started = True
|
|
return
|
|
_banner_data_prefetch_started = True
|
|
_daemon("banner-data-prefetch", lambda: [_quiet(warm) for warm in (
|
|
get_git_banner_state, get_latest_release_tag, get_available_skills)])
|
|
|
|
|
|
def get_update_result(timeout: float = 0.5) -> Optional[int]:
|
|
"""Get result of prefetched check. Returns None if not ready."""
|
|
_update_check_done.wait(timeout=timeout)
|
|
return _update_result
|
|
|
|
|
|
def _format_update_notice(behind: int) -> str:
|
|
"""Render the update warning line for a non-zero ``behind`` result."""
|
|
from hermes_cli.config import get_managed_update_command, recommended_update_command
|
|
if behind > 0:
|
|
return (
|
|
f"[bold yellow]⚠ {behind} {_plural(behind, 'commit')} behind[/]"
|
|
f"[dim yellow] — run [bold]{recommended_update_command()}[/bold] to update[/]")
|
|
# UPDATE_AVAILABLE_NO_COUNT (nix): an update exists but we don't know by how much, nor how
|
|
# the user installed (nix run, profile, system flake, home-manager).
|
|
managed_cmd = get_managed_update_command()
|
|
suffix = f"[dim yellow] — run [bold]{managed_cmd}[/bold][/]" if managed_cmd else ""
|
|
return f"[bold yellow]⚠ update available[/]{suffix}"
|
|
|
|
|
|
_deferred_update_notice_started = False
|
|
|
|
|
|
def _render_markup_to_ansi(markup: str) -> str:
|
|
"""Rich markup → ANSI string, for output that must go through prompt_toolkit's renderer.
|
|
|
|
Under ``patch_stdout`` (the interactive CLI), a plain ``Console.print`` writes ESC bytes into
|
|
the StdoutProxy, which sanitizes them into visible ``?[1;33m…`` artifacts (#83969).
|
|
"""
|
|
from io import StringIO
|
|
from rich.console import Console as _Console
|
|
buf = StringIO()
|
|
_Console(file=buf, force_terminal=True, color_system="truecolor", highlight=False).print(markup)
|
|
return buf.getvalue().rstrip("\n")
|
|
|
|
|
|
def _defer_update_notice(max_wait: float = 30.0) -> None:
|
|
"""Print the update warning once the prefetched check completes (at most once per process).
|
|
|
|
Used when the banner rendered before the update prefetch finished so startup never blocks on
|
|
git/network. The notice lands after prompt_toolkit owns the terminal, so it is routed through
|
|
``cprint`` (prompt_toolkit's renderer prints above a running application from any thread).
|
|
"""
|
|
global _deferred_update_notice_started
|
|
if _deferred_update_notice_started:
|
|
return
|
|
_deferred_update_notice_started = True
|
|
|
|
def _wait_and_print() -> None:
|
|
if _update_check_done.wait(timeout=max_wait) and _update_result:
|
|
cprint(_render_markup_to_ansi(_format_update_notice(_update_result)))
|
|
_daemon("update-notice", _wait_and_print) # never break the session over an update notice
|
|
|
|
|
|
# === Welcome banner ===
|
|
|
|
def _plural(n: int, word: str) -> str:
|
|
return word if n == 1 else f"{word}s"
|
|
|
|
|
|
def _format_context_length(tokens: int) -> str:
|
|
"""Format a token count for display (e.g. 128000 → '128K', 1048576 → '1M')."""
|
|
for unit, div in (("M", 1_000_000), ("K", 1_000)):
|
|
if tokens >= div:
|
|
val = tokens / div
|
|
rounded = round(val)
|
|
return f"{rounded}{unit}" if abs(val - rounded) < 0.05 else f"{val:.1f}{unit}"
|
|
return str(tokens)
|
|
|
|
|
|
def _display_toolset_name(toolset_name: str) -> str:
|
|
"""Normalize internal/legacy toolset identifiers for banner display."""
|
|
return toolset_name.removesuffix("_tools") if toolset_name else "unknown"
|
|
|
|
|
|
def _short_label(name: str) -> str:
|
|
"""Truncate a model/preset slug to fit the banner's left column."""
|
|
return name[:25] + "..." if len(name) > 28 else name
|
|
|
|
|
|
# === Banner snapshot — warm-launch fast path ===
|
|
# The tool panel needs the full tool registry (~0.5-0.9s cold, the largest chunk of time-to-
|
|
# banner). The list is a pure function of (config.yaml, .env, code checkout, enabled toolsets),
|
|
# so the rendered inputs are snapshotted to disk and replayed when the fingerprint matches. The
|
|
# agent's REAL tool list is still computed fresh at first message; the snapshot only feeds the
|
|
# cosmetic panel, and a background refresh (cli.show_banner) re-verifies it right after render.
|
|
|
|
_BANNER_SNAPSHOT_VERSION = 1
|
|
|
|
|
|
def _banner_snapshot_path() -> Path:
|
|
return get_hermes_home() / "cache" / "banner_snapshot.json"
|
|
|
|
|
|
def banner_snapshot_fingerprint() -> Optional[str]:
|
|
"""Fingerprint the inputs the banner tool panel depends on."""
|
|
import hashlib
|
|
def _inputs():
|
|
from hermes_cli.config import get_config_path
|
|
return (get_config_path(), get_hermes_home() / ".env")
|
|
paths = _quiet(_inputs)
|
|
if paths is None:
|
|
return None
|
|
parts = [f"v{_BANNER_SNAPSHOT_VERSION}"]
|
|
for p in paths:
|
|
st = _quiet(p.stat)
|
|
parts.append(f"{p.name}:{st.st_mtime_ns}:{st.st_size}" if st else f"{p.name}:absent")
|
|
# Code checkout: commit when known, otherwise its derived version.
|
|
version_info = get_version_info()
|
|
parts.append(version_info.commit or version_info.derived_version)
|
|
state = get_git_banner_state()
|
|
if state:
|
|
parts.append(str(state.get("local", "")))
|
|
return hashlib.sha256("|".join(parts).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def load_banner_snapshot(enabled_toolsets: List[str] = None) -> Optional[Dict[str, Any]]:
|
|
"""Return the stored banner snapshot when its fingerprint is current."""
|
|
blob = _quiet(lambda: json.loads(_banner_snapshot_path().read_text(encoding="utf-8-sig")))
|
|
if not isinstance(blob, dict):
|
|
return None
|
|
fp = banner_snapshot_fingerprint()
|
|
if (not fp or blob.get("fingerprint") != fp
|
|
or blob.get("enabled_toolsets") != sorted(enabled_toolsets or [])
|
|
or not isinstance(blob.get("tools"), list)
|
|
or not all(isinstance(blob.get(k), dict)
|
|
for k in ("toolset_map", "availability", "skills_by_category"))):
|
|
return None
|
|
return blob
|
|
|
|
|
|
def save_banner_snapshot(tools: List[dict], enabled_toolsets: List[str], availability: Dict[str, Any],
|
|
toolset_map: Dict[str, str]) -> None:
|
|
"""Persist the banner tool panel inputs for next launch (best-effort)."""
|
|
fp = banner_snapshot_fingerprint()
|
|
if not fp:
|
|
return
|
|
payload = {
|
|
"fingerprint": fp,
|
|
"enabled_toolsets": sorted(enabled_toolsets or []),
|
|
"tools": [{"function": {"name": t["function"]["name"]}}
|
|
for t in tools if isinstance(t, dict) and t.get("function", {}).get("name")],
|
|
"toolset_map": toolset_map,
|
|
"availability": {
|
|
"unavailable_toolsets": availability.get("unavailable_toolsets", []),
|
|
**{k: list(availability.get(k, [])) for k in ("lazy_tools", "disabled_tools")}},
|
|
"skills_by_category": get_available_skills(),
|
|
}
|
|
|
|
def _write():
|
|
from utils import atomic_json_write
|
|
atomic_json_write(_banner_snapshot_path(), payload, indent=None, mode=0o600)
|
|
_quiet(_write)
|
|
|
|
|
|
def compute_toolset_availability(enabled_toolsets: List[str] = None) -> Dict[str, Any]:
|
|
"""Compute ``{"unavailable_toolsets", "lazy_tools", "disabled_tools"}`` for the banner.
|
|
|
|
Split out so the result can be snapshotted and replayed without importing ``model_tools``.
|
|
"""
|
|
from model_tools import check_tool_availability, TOOLSET_REQUIREMENTS
|
|
enabled_toolsets = enabled_toolsets or []
|
|
_, unavailable_toolsets = check_tool_availability(quiet=True)
|
|
# The availability check walks the GLOBAL registry, so it includes toolsets outside this
|
|
# agent's platform set (e.g. `discord` on a CLI session) which must never surface in
|
|
# "Available Tools". Restrict to enabled toolsets; an enabled toolset with unmet deps
|
|
# legitimately shows as disabled/lazy below.
|
|
_enabled_ts = {str(t) for t in enabled_toolsets}
|
|
if _enabled_ts:
|
|
unavailable_toolsets = [
|
|
item for item in unavailable_toolsets if str(item.get("id", item.get("name", ""))) in _enabled_ts]
|
|
# Toolsets with a check_fn are lazy-initialized (e.g. honcho): unavailable at banner time
|
|
# because the check hasn't run yet, but not misconfigured.
|
|
lazy_tools, disabled_tools = set(), set()
|
|
for item in unavailable_toolsets:
|
|
is_lazy = TOOLSET_REQUIREMENTS.get(item.get("name", ""), {}).get("check_fn")
|
|
(lazy_tools if is_lazy else disabled_tools).update(item.get("tools", []))
|
|
return {"unavailable_toolsets": unavailable_toolsets, "lazy_tools": sorted(lazy_tools),
|
|
"disabled_tools": sorted(disabled_tools)}
|
|
|
|
|
|
def _mcp_server_line(srv: dict, *, dim: str, text: str) -> str:
|
|
"""One banner line for an MCP server status entry."""
|
|
name, transport = srv["name"], srv["transport"]
|
|
if srv["connected"]:
|
|
return f"[dim {dim}]{name}[/] [{text}]({transport})[/] [dim {dim}]—[/] [{text}]{srv['tools']} tool(s)[/]"
|
|
# Needs srv['tools'], so it cannot live in the suffix dict below. A registered but unspawned
|
|
# server has callable tools; falling through to the red "failed" line misreports a working setup.
|
|
if srv.get("status") == "lazy":
|
|
return (f"[dim {dim}]{name}[/] [{text}]({transport})[/] [dim {dim}]—[/] "
|
|
f"[{text}]{srv['tools']} tool(s)[/] [dim {dim}](lazy, starts on first use)[/]")
|
|
status = "disabled" if srv.get("disabled") else srv.get("status")
|
|
suffix = {"disabled": f"[dim {dim}]— disabled[/]", "connecting": "[yellow]— connecting[/]",
|
|
"configured": f"[dim {dim}]— configured[/]"}.get(status)
|
|
if suffix is not None:
|
|
return f"[dim {dim}]{name}[/] [dim]({transport})[/] {suffix}"
|
|
return _mcp_failed_line(name, transport, srv.get("error"))
|
|
|
|
|
|
def _mcp_failed_line(name: str, transport: str, error: Optional[str]) -> str:
|
|
"""Failed MCP connect: the short reason (already humanised by ``_format_connect_error``) and the
|
|
exact next command, so 'failed' is never the whole story."""
|
|
from rich.markup import escape
|
|
reason = escape(" ".join(str(error or "").split())[:120]) or "no details recorded"
|
|
next_cmd = (f"hermes mcp login {name}" if re.search(r"\b401\b|unauthori[sz]ed", reason, re.I)
|
|
else f"hermes mcp test {name}")
|
|
return (f"[red]{name}[/] [dim]({transport})[/] [red]— could not connect:[/] {reason} "
|
|
f"[dim]— run `{next_cmd}`[/]")
|
|
|
|
|
|
def _truncate_tool_names(tool_names: List[str]) -> List[Optional[str]]:
|
|
"""Cut a toolset's tool list to ~42 columns; ``None`` marks the elided tail."""
|
|
if len(", ".join(tool_names)) <= 45:
|
|
return list(tool_names)
|
|
short_names: List[Optional[str]] = []
|
|
length = 0
|
|
for name in tool_names:
|
|
if length + len(name) + 2 > 42:
|
|
short_names.append(None)
|
|
break
|
|
short_names.append(name)
|
|
length += len(name) + 2
|
|
return short_names
|
|
|
|
|
|
def _pack_skill_names(skill_names: List[str], avail: int) -> str:
|
|
"""Join skill names into ``avail`` columns, ending with ``+N more`` when they don't all fit."""
|
|
parts: List[str] = []
|
|
length = 0
|
|
for i, name in enumerate(skill_names):
|
|
needed = (2 if parts else 0) + len(name)
|
|
after = len(skill_names) - (i + 1) # indicator size IF we add this skill then stop
|
|
ind_len = len(f", +{after} more") if after > 0 else 0
|
|
if parts and length + needed + ind_len > avail:
|
|
parts.append(f"+{len(skill_names) - len(parts)} more")
|
|
break
|
|
parts.append(name)
|
|
length += needed
|
|
return ", ".join(parts)
|
|
|
|
|
|
def _moa_aggregator_label(preset_name: str) -> str:
|
|
"""Short aggregator-model label for a MoA preset ("" when the preset has none)."""
|
|
from hermes_cli.config import load_config
|
|
from hermes_cli.moa_config import normalize_moa_config
|
|
preset = normalize_moa_config(load_config().get("moa") or {}).get("presets", {}).get(preset_name)
|
|
model = str(((preset or {}).get("aggregator") or {}).get("model") or "")
|
|
return model.split("/")[-1]
|
|
|
|
|
|
def _mcp_configured() -> bool:
|
|
"""Cheap probe: does config.yaml or the persisted plugin key cache name any MCP server?
|
|
|
|
The full ``get_mcp_status()`` path resolves portable plugin MCP servers, which JOINS the in-flight
|
|
background plugin discovery (~100ms on the startup path), so skip it when nothing is configured.
|
|
When either probe can't tell, take the full path.
|
|
"""
|
|
def _native():
|
|
from hermes_cli.config import load_config
|
|
return bool((load_config() or {}).get("mcp_servers"))
|
|
|
|
def _portable():
|
|
from hermes_cli.plugins import get_portable_mcp_server_names_nowait
|
|
return bool(get_portable_mcp_server_names_nowait())
|
|
return _quiet(_native, True) or _quiet(_portable, True)
|
|
|
|
|
|
def _probe_mcp_status() -> list:
|
|
from tools.mcp_tool_discovery import get_mcp_status
|
|
return get_mcp_status()
|
|
|
|
|
|
def _codex_runtime_active() -> bool:
|
|
"""True when the codex_app_server runtime is active (tool counts then live inside codex)."""
|
|
from hermes_cli.codex_runtime_switch import get_current_runtime
|
|
from hermes_cli.config import load_config
|
|
return get_current_runtime(load_config()) == "codex_app_server"
|
|
|
|
|
|
def _active_profile_name() -> Optional[str]:
|
|
from hermes_cli.profiles import get_active_profile_name
|
|
return get_active_profile_name()
|
|
|
|
|
|
def _route_model_for_banner(provider: Any) -> str:
|
|
"""The model the resolved route will actually serve when config names none: today only the Nous
|
|
free tier (welcome host -> ``nous/welcome``). Read from the boot record and local auth state;
|
|
no network. Empty when nothing resolves, so the caller keeps its "no model configured" line."""
|
|
if (provider or "auto").strip().lower() not in ("auto", "nous"):
|
|
return ""
|
|
from hermes_cli.anon_auth import GUEST_MODEL, guest_carries_inference
|
|
return GUEST_MODEL if guest_carries_inference() else ""
|
|
|
|
|
|
def _banner_left_lines(model: str, cwd: str, session_id, context_length, provider, *, accent: str, dim: str,
|
|
context_pinned: bool = False) -> list:
|
|
"""Model / cwd / session lines under the hero art. ``context_pinned`` marks a
|
|
``model.context_length`` pin so the user can tell it apart from provider metadata (#66168)."""
|
|
def _dim_sep(label: str) -> str:
|
|
return f" [dim {dim}]·[/] [dim {dim}]{label}[/]"
|
|
lines = []
|
|
pin = " (pinned)" if context_pinned else ""
|
|
ctx_str = _dim_sep(f"{_format_context_length(context_length)} context{pin}") if context_length else ""
|
|
nous_str = _dim_sep("Nous Research")
|
|
if not (model or "").strip():
|
|
# Credentials resolve lazily on the first message; the banner prints first. Ask the route
|
|
# the same question so a fresh free-tier install shows its model, not a red "unconfigured".
|
|
model = _quiet(lambda: _route_model_for_banner(provider), "") or model
|
|
if (provider or "").strip().lower() == "moa":
|
|
# MoA virtual provider: ``model`` is a preset name; show it with its aggregator.
|
|
agg_label = _quiet(lambda: _moa_aggregator_label(model), "")
|
|
agg_str = _dim_sep(f"agg {agg_label}") if agg_label else ""
|
|
lines.append(f"[{accent}]MoA: {_short_label(model)}[/]{agg_str}{ctx_str}{nous_str}")
|
|
elif not (model or "").strip() or (model or "").strip().lower() == "unknown":
|
|
# Unconfigured install: the clearest place to say what is wrong and how to fix it.
|
|
lines.append(f"[bold red]no model configured[/] [dim {dim}]— run /model or hermes setup[/]")
|
|
else:
|
|
model_short = model.split("/")[-1].removesuffix(".gguf")
|
|
lines.append(f"[{accent}]{_short_label(model_short)}[/]{ctx_str}{nous_str}")
|
|
if os.getenv("HERMES_YOLO_MODE"):
|
|
lines.append(f"[bold red]⚠ YOLO mode[/] [dim {dim}]— all approval prompts bypassed[/]")
|
|
lines.append(f"[dim {dim}]{cwd}[/]")
|
|
if session_id:
|
|
lines.append(f"[dim {_skin_color('session_border', '#8B8682')}]Session: {session_id}[/]")
|
|
return lines
|
|
|
|
|
|
def _banner_tool_lines(
|
|
tools: list, unavailable_toolsets: list, get_toolset_for_tool, *,
|
|
lazy_tools: set, disabled_tools: set, accent: str, dim: str, text: str) -> list:
|
|
""""Available Tools" section: up to 8 toolsets, each truncated to ~42 columns."""
|
|
lines = [f"[bold {accent}]Available Tools[/]"]
|
|
toolsets_dict: Dict[str, list] = {}
|
|
for tool in tools:
|
|
tool_name = tool["function"]["name"]
|
|
toolset = _display_toolset_name(get_toolset_for_tool(tool_name) or "other")
|
|
toolsets_dict.setdefault(toolset, []).append(tool_name)
|
|
for item in unavailable_toolsets:
|
|
names = toolsets_dict.setdefault(_display_toolset_name(item.get("id", item.get("name", "unknown"))), [])
|
|
for tool_name in item.get("tools", []):
|
|
if tool_name not in names:
|
|
names.append(tool_name)
|
|
|
|
def _color_tool(name: Optional[str]) -> str:
|
|
if name is None: # truncation marker
|
|
return "[dim]...[/]"
|
|
color = "red" if name in disabled_tools else "yellow" if name in lazy_tools else text
|
|
return f"[{color}]{name}[/]"
|
|
sorted_toolsets = sorted(toolsets_dict.keys())
|
|
for toolset in sorted_toolsets[:8]:
|
|
tool_names = _truncate_tool_names(sorted(toolsets_dict[toolset]))
|
|
lines.append(f"[dim {dim}]{toolset}:[/] {', '.join(_color_tool(n) for n in tool_names)}")
|
|
if len(sorted_toolsets) > 8:
|
|
lines.append(f"[dim {dim}](and {len(sorted_toolsets) - 8} more toolsets...)[/]")
|
|
return lines
|
|
|
|
|
|
def _banner_skill_lines(skills_by_category: Dict[str, List[str]], skills_enabled: bool, *, dim: str, text: str) -> list:
|
|
""""Available Skills" body, sized to ~60% of the terminal width (the right grid column)."""
|
|
if not skills_enabled:
|
|
return [f"[dim {dim}]Skills toolset disabled[/]"]
|
|
if not skills_by_category:
|
|
return [f"[dim {dim}]No skills installed[/]"]
|
|
right_col_width = max(int(shutil.get_terminal_size().columns * 0.6) - 10, 30)
|
|
lines = []
|
|
for category in sorted(skills_by_category.keys()):
|
|
# Account for the "category: " prefix.
|
|
skills_str = _pack_skill_names(sorted(skills_by_category[category]), max(right_col_width - len(category) - 2, 20))
|
|
lines.append(f"[dim {dim}]{category}:[/] [{text}]{skills_str}[/]")
|
|
return lines
|
|
|
|
|
|
def build_welcome_banner(
|
|
console: "Console", model: str, cwd: str, tools: List[dict] = None, enabled_toolsets: List[str] = None,
|
|
session_id: str = None, get_toolset_for_tool=None, context_length: int = None, provider: str = None,
|
|
availability: Dict[str, Any] = None, skills_by_category: Dict[str, List[str]] = None,
|
|
context_pinned: bool = False,
|
|
):
|
|
"""Build and print a welcome banner with caduceus on left and info on right.
|
|
|
|
When ``provider == "moa"``, ``model`` is a MoA preset name and the aggregator is rendered.
|
|
Passing a precomputed ``availability`` together with ``get_toolset_for_tool`` avoids any
|
|
``model_tools`` import (banner snapshot replay).
|
|
"""
|
|
from rich.panel import Panel
|
|
from rich.table import Table
|
|
if get_toolset_for_tool is None:
|
|
from model_tools import get_toolset_for_tool
|
|
tools = tools or []
|
|
enabled_toolsets = enabled_toolsets or []
|
|
if availability is None:
|
|
availability = compute_toolset_availability(enabled_toolsets)
|
|
_enabled_ts = {str(t) for t in enabled_toolsets}
|
|
# Resolve skin colors once for the entire banner
|
|
accent = _skin_color("banner_accent", "#FFBF00")
|
|
dim = _skin_color("banner_dim", "#B8860B")
|
|
text = _skin_color("banner_text", "#FFF8DC")
|
|
# Use skin's custom caduceus art if provided
|
|
_bskin = _quiet(_active_skin)
|
|
left_lines = ["", getattr(_bskin, "banner_hero", None) or HERMES_CADUCEUS, ""]
|
|
left_lines += _banner_left_lines(model, cwd, session_id, context_length, provider, accent=accent, dim=dim,
|
|
context_pinned=context_pinned)
|
|
right_lines = _banner_tool_lines(
|
|
tools, availability.get("unavailable_toolsets", []), get_toolset_for_tool,
|
|
lazy_tools=set(availability.get("lazy_tools", [])), disabled_tools=set(availability.get("disabled_tools", [])),
|
|
accent=accent, dim=dim, text=text)
|
|
# MCP Servers section (only if configured) — see ``_mcp_configured`` for why the cheap probe.
|
|
mcp_status = _quiet(_probe_mcp_status, []) if _mcp_configured() else []
|
|
if mcp_status:
|
|
right_lines += ["", f"[bold {accent}]MCP Servers[/]"]
|
|
right_lines.extend(_mcp_server_line(srv, dim=dim, text=text) for srv in mcp_status)
|
|
right_lines += ["", f"[bold {accent}]Available Skills[/]"]
|
|
# The skills catalog is only reachable when the `skills` toolset is enabled (skill_view /
|
|
# skill_manage). When disabled (Blank Slate) the agent cannot load any skill, so advertising
|
|
# the on-disk catalog would be misleading — reflect the real state.
|
|
_skills_enabled = (not _enabled_ts) or ("skills" in _enabled_ts)
|
|
if not _skills_enabled:
|
|
skills_by_category = {}
|
|
elif skills_by_category is None:
|
|
skills_by_category = get_available_skills()
|
|
total_skills = sum(len(s) for s in skills_by_category.values())
|
|
right_lines += _banner_skill_lines(skills_by_category, _skills_enabled, dim=dim, text=text)
|
|
right_lines.append("")
|
|
mcp_connected = sum(1 for s in mcp_status if s["connected"])
|
|
summary_parts = [f"{len(tools)} tools", f"{total_skills} skills"]
|
|
if mcp_connected:
|
|
summary_parts.append(f"{mcp_connected} MCP servers")
|
|
summary_parts.append("/help for commands")
|
|
# Flag the codex_app_server runtime so users understand why tool counts may not match what's
|
|
# reachable (codex builds its own tool list inside the spawned subprocess).
|
|
if _quiet(_codex_runtime_active, False):
|
|
right_lines.append(f"[bold {accent}]Runtime:[/] [{text}]codex app-server[/] "
|
|
f"[dim {dim}](terminal/file ops/MCP run inside codex)[/]")
|
|
# Show active profile name when not 'default'. Never break the banner over a profiles.py bug.
|
|
_profile_name = _quiet(_active_profile_name)
|
|
if _profile_name and _profile_name != "default":
|
|
right_lines.append(f"[bold {accent}]Profile:[/] [{text}]{_profile_name}[/]")
|
|
right_lines.append(f"[dim {dim}]{' · '.join(summary_parts)}[/]")
|
|
# Update check — NEVER block the banner on it: the prefetch does git/network work that rarely
|
|
# finishes before render, so a blocking wait adds its full timeout to every startup. If not
|
|
# ready, a daemon thread prints the same notice above the prompt when it lands.
|
|
def _update_line():
|
|
behind = get_update_result(timeout=0.05)
|
|
if behind is None and not _update_check_done.is_set():
|
|
_defer_update_notice()
|
|
elif behind is not None and behind != 0:
|
|
right_lines.append(_format_update_notice(behind))
|
|
_quiet(_update_line) # Never break the banner over an update check
|
|
layout_table = Table.grid(padding=(0, 2))
|
|
layout_table.add_column("left", justify="left")
|
|
layout_table.add_column("right", justify="left")
|
|
layout_table.add_row("\n".join(left_lines), "\n".join(right_lines))
|
|
version_label = format_banner_version_label()
|
|
release_info = get_latest_release_tag()
|
|
if release_info:
|
|
version_label = f"[link={release_info[1]}]{version_label}[/link]"
|
|
outer_panel = Panel(
|
|
layout_table, title=f"[bold {_skin_color('banner_title', '#FFD700')}]{version_label}[/]",
|
|
border_style=_skin_color("banner_border", "#CD7F32"), padding=(0, 2))
|
|
console.print()
|
|
if shutil.get_terminal_size().columns >= 95:
|
|
console.print(getattr(_bskin, "banner_logo", None) or HERMES_AGENT_LOGO)
|
|
console.print()
|
|
console.print(outer_panel)
|