ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-16-D' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:57:45 -07:00
parent 1006954d73 3c9390a66d
commit 601488dca7
5 changed files with 715 additions and 1425 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/__init__.py
+10 -16
View File
@@ -10,9 +10,9 @@ __release_date__ = "2026.8.31"
def _ensure_utf8():
"""Force UTF-8 stdout/stderr to prevent UnicodeEncodeError crashes.
The CLI prints box-drawing characters (┌│├└─) and the ⚕ glyph in the setup wizard, doctor, and
status banners. Encoding those under a non-UTF-8 codec raises an unhandled UnicodeEncodeError
that crashes the command before it can even start — e.g. `hermes setup` on a fresh Pi.
The CLI prints box-drawing characters and the ⚕ glyph in the setup wizard, doctor, and status
banners; under a non-UTF-8 codec that raises before the command can even start (e.g.
`hermes setup` on a fresh Pi).
"""
repaired = False
@@ -25,30 +25,24 @@ def _ensure_utf8():
if encoding == "utf8":
continue
# Preferred: reconfigure the existing TextIOWrapper in place. This
# preserves object identity so any code already holding a reference
# to the old sys.stdout benefits from the repair too.
# Preferred: reconfigure in place, preserving object identity so code already holding
# a reference to the old sys.stdout benefits from the repair too.
reconfigure = getattr(stream, "reconfigure", None)
if callable(reconfigure):
reconfigure(encoding="utf-8", errors="replace")
repaired = True
continue
# Fallback: reopen the underlying file descriptor as UTF-8. Used
# for streams that don't expose reconfigure() (e.g. some wrapped
# or replaced streams). closefd=False keeps the original fd open.
new_stream = open(
stream.fileno(), "w", encoding="utf-8",
errors="replace", buffering=1, closefd=False,
)
# Fallback for streams without reconfigure(): reopen the fd as UTF-8 (closefd=False
# keeps the original fd open).
new_stream = open(stream.fileno(), "w", encoding="utf-8", errors="replace", buffering=1, closefd=False)
setattr(sys, stream_name, new_stream)
repaired = True
except (AttributeError, OSError, ValueError):
pass
# Only nudge child processes toward UTF-8 when we actually detected a
# non-UTF-8 locale. On a healthy UTF-8 host children inherit UTF-8 from the
# locale already, so leave the environment untouched (minimal footprint).
# Only nudge child processes toward UTF-8 when a non-UTF-8 locale was actually detected; on a
# healthy UTF-8 host children inherit it from the locale already.
if repaired:
os.environ.setdefault("PYTHONUTF8", "1")
os.environ.setdefault("PYTHONIOENCODING", "utf-8")
hermes_cli/archive_safe.py
+11 -23
View File
@@ -10,7 +10,7 @@ from pathlib import Path, PurePosixPath, PureWindowsPath
def normalize_archive_parts(member_name: str) -> list[str]:
"""Return safe path parts for an archive member, or raise.
"""Return safe path parts for an archive member, or raise ``ValueError``.
Rejects absolute paths (POSIX and Windows, including drive letters), empty names, and any ``..``
component. Backslashes are folded to ``/`` first so a Windows-authored archive can't smuggle a
@@ -20,16 +20,11 @@ def normalize_archive_parts(member_name: str) -> list[str]:
posix_path = PurePosixPath(normalized_name)
windows_path = PureWindowsPath(member_name)
if (
not normalized_name
or posix_path.is_absolute()
or windows_path.is_absolute()
or windows_path.drive
):
if not normalized_name or posix_path.is_absolute() or windows_path.is_absolute() or windows_path.drive:
raise ValueError(f"Unsafe archive member path: {member_name}")
parts = [part for part in posix_path.parts if part not in {"", "."}]
if not parts or any(part == ".." for part in parts):
if not parts or ".." in parts:
raise ValueError(f"Unsafe archive member path: {member_name}")
return parts
@@ -37,14 +32,11 @@ def normalize_archive_parts(member_name: str) -> list[str]:
def make_targz(base: str, root_dir: str, base_dir: str) -> str:
"""Create ``<base>.tar.gz`` of ``root_dir/base_dir`` in GNU tar format.
Writes to a sibling temp file and renames onto ``archive_path`` only after the archive is fully
written.
Writes to a sibling temp file and renames onto the archive path only after it is fully written.
"""
archive_path = f"{base}.tar.gz"
dest_dir = os.path.dirname(archive_path) or "."
fd, tmp_path = tempfile.mkstemp(
dir=dest_dir, prefix=".archive_", suffix=".tar.gz.tmp"
)
fd, tmp_path = tempfile.mkstemp(dir=dest_dir, prefix=".archive_", suffix=".tar.gz.tmp")
try:
with os.fdopen(fd, "wb") as f:
with tarfile.open(fileobj=f, mode="w:gz", format=tarfile.GNU_FORMAT) as tf:
@@ -68,17 +60,14 @@ def safe_extract_targz(archive: Path, destination: Path) -> None:
"""
with tarfile.open(archive, "r:gz") as tf:
for member in tf.getmembers():
parts = normalize_archive_parts(member.name)
target = destination.joinpath(*parts)
target = destination.joinpath(*normalize_archive_parts(member.name))
if member.isdir():
target.mkdir(parents=True, exist_ok=True)
continue
if not member.isfile():
raise ValueError(
f"Unsupported archive member type: {member.name}"
)
raise ValueError(f"Unsupported archive member type: {member.name}")
target.parent.mkdir(parents=True, exist_ok=True)
extracted = tf.extractfile(member)
@@ -98,8 +87,8 @@ def archive_root_dirs(archive: Path) -> set[str]:
"""Return the archive's top-level directory names.
Transfer archives carry exactly one root directory, which names the thing being imported.
Inspecting the archive before extraction lets the caller resolve the target name (and refuse a
malformed archive) without first mutating a live tree.
Inspecting before extraction lets the caller resolve the target name (and refuse a malformed
archive) without first mutating a live tree.
"""
with tarfile.open(archive, "r:gz") as tf:
return {
@@ -111,11 +100,10 @@ def archive_root_dirs(archive: Path) -> set[str]:
def copy_regular_files(src: Path, dst: Path) -> int:
"""Copy the regular files under ``src`` into ``dst``, skipping symlinks.
"""Copy the regular files under ``src`` into ``dst``, skipping symlinks; return the count.
Used on the *export* side so a symlink planted in an attachments or logs tree can't pull an
arbitrary file into the archive. Returns the number of files copied; a missing ``src`` copies
nothing.
arbitrary file into the archive. A missing ``src`` copies nothing.
"""
if not src.is_dir():
return 0
hermes_cli/backup.py
+463 -1016
View File
File diff suppressed because it is too large Load Diff
hermes_cli/bang_shell.py
+28 -49
View File
@@ -1,12 +1,9 @@
"""``!<command>`` shell mode for the interactive CLI.
Typing ``!git status`` at the composer runs the command directly in the session's working directory.
The model is never invoked: no user message, no assistant message, no tool result enters the
conversation history, so a bang command costs zero tokens and cannot perturb role alternation or the
prompt cache.
CLI-only by design: gateway/API/cron sessions have their own shells and no composer, so
:func:`bang_shell_enabled` gates the feature off there.
``!git status`` at the composer runs directly in the session's working directory. The model is
never invoked — no user/assistant message or tool result enters history — so a bang command costs
zero tokens and cannot perturb role alternation or the prompt cache. CLI-only by design:
gateway/API/cron sessions have their own shells and no composer (:func:`bang_shell_enabled`).
"""
from __future__ import annotations
@@ -17,27 +14,25 @@ from typing import Optional
USAGE_HINT = "Usage: !<command> — run a shell command without spending a model turn (e.g. !git status)"
# Bang commands are interactive convenience, not agent work. Keep the ceiling
# well under the terminal tool's foreground cap: a user watching output can
# Ctrl+C, and an accidental `!sleep 999` should not wedge the composer.
# Interactive convenience, not agent work: keep the ceiling well under the terminal tool's foreground
# cap so an accidental `!sleep 999` cannot wedge the composer (the user can Ctrl+C anyway).
DEFAULT_TIMEOUT = 120
def is_bang_command(text: Optional[str]) -> bool:
"""Return True when *text* is a ``!`` shell-mode submission.
"""True when *text* is a ``!`` shell-mode submission.
Only a leading ``!`` (after surrounding whitespace) counts. A line that merely *contains* ``!``
mid-text (``fix the bug!``, ``echo hi!``) is an ordinary prompt and must reach the agent
untouched.
Only a leading ``!`` (after surrounding whitespace) counts; ``fix the bug!`` is an ordinary
prompt and must reach the agent untouched.
"""
return isinstance(text, str) and text.strip().startswith("!")
def parse_bang_command(text: str) -> str:
"""Return the shell command inside a bang submission (``""`` when bare).
"""The shell command inside a bang submission (``""`` when bare).
``! ls -la`` -> ``ls -la``; ``!!`` -> ``!`` -- a literal second bang is part of the command
(e.g. history expansion for the user's shell), not a Hermes token.
``! ls -la`` -> ``ls -la``; ``!!`` -> ``!`` — a literal second bang belongs to the user's shell
(history expansion), not to Hermes.
"""
if not is_bang_command(text):
return ""
@@ -65,11 +60,10 @@ def bang_shell_enabled() -> bool:
def resolve_bang_cwd(session_key: Optional[str] = None) -> Optional[str]:
"""Return the directory a bang command should run in.
"""The directory a bang command should run in.
Mirrors the terminal tool's resolution order so ``!pwd`` matches where the agent's own commands
land: the session's recorded ``cd`` state first (``terminal_tool.get_session_cwd``, updated
after every agent command), then the configured ``TERMINAL_CWD``/backend default.
Mirrors the terminal tool's order so ``!pwd`` matches the agent's own commands: the session's
recorded ``cd`` state first, then the configured ``TERMINAL_CWD``/backend default.
"""
try:
from tools.terminal_tool import _get_env_config, get_session_cwd
@@ -82,30 +76,26 @@ def resolve_bang_cwd(session_key: Optional[str] = None) -> Optional[str]:
def check_bang_approval(command: str) -> dict:
"""Run *command* through the terminal tool's approval gate.
Reuses ``tools.terminal_tool._check_all_guards`` — the exact function ``terminal_tool()`` calls
before executing anything — so the hardline blocklist, user deny rules, tirith findings, and the
interactive dangerous-command prompt all apply to user-typed bang commands too.
Returns the gate's decision dict (``{"approved": bool, "message": ...}``). Falls back to
*approved* only when the gate itself cannot be imported, which would mean a broken install
rather than a policy decision.
Reuses ``tools.terminal_tool._check_all_guards`` — exactly what ``terminal_tool()`` calls — so the
hardline blocklist, user deny rules, tirith findings, and the dangerous-command prompt all apply
to user-typed bang commands too. Returns the gate's ``{"approved": bool, "message": ...}``;
falls back to *approved* only when the gate itself cannot be imported (a broken install, not a
policy decision).
"""
try:
from tools.terminal_tool import _check_all_guards
except Exception:
return {"approved": True, "message": None}
# env_type mirrors the terminal tool: bang commands always run locally in
# the CLI process, never inside a remote/sandbox backend.
# Bang commands always run locally in the CLI process, never inside a remote/sandbox backend.
return _check_all_guards(command, "local", has_host_access=False)
def _bang_env() -> dict:
"""Environment for a bang command, with Hermes-managed secrets filtered.
"""Environment for a bang command with Hermes-managed secrets filtered.
The CLI process holds every provider API key in ``os.environ``; a bang command is user-typed
but may still run a third-party script, so reuse the sanitizer ``quick_commands`` and the
local terminal backend use rather than hand the whole keyring to an arbitrary subprocess.
The CLI process holds every provider API key; a user-typed command may still run a third-party
script, so reuse the sanitizer ``quick_commands`` and the local terminal backend use.
"""
try:
from tools.environments.local import _sanitize_subprocess_env
@@ -115,19 +105,10 @@ def _bang_env() -> dict:
return os.environ.copy()
def run_bang_command(
command: str,
*,
cwd: Optional[str] = None,
timeout: int = DEFAULT_TIMEOUT,
writer=None,
) -> int:
"""Execute *command* and stream its output, returning the exit code.
def run_bang_command(command: str, *, cwd: Optional[str] = None, timeout: int = DEFAULT_TIMEOUT, writer=None) -> int:
"""Execute *command*, streaming merged stdout/stderr through *writer* (default ``print``); return the exit code.
stdout and stderr are merged and written through *writer* (defaults to ``print``) as they
arrive, so long-running commands show progress instead of buffering to the end. Nothing is
returned to a caller for insertion into conversation history — the output exists only on the
user's terminal.
Output exists only on the user's terminal — nothing is returned for insertion into history.
"""
emit = writer or (lambda line: print(line, end="" if line.endswith("\n") else "\n"))
@@ -143,8 +124,7 @@ def run_bang_command(
creationflags = 0
try:
# shell=True is intentional and matches quick_commands: this is a
# command the human typed into their own composer, not model output.
# shell=True is intentional (matches quick_commands): the human typed this, not the model.
proc = subprocess.Popen(
command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, encoding="utf-8", errors="replace",
@@ -176,4 +156,3 @@ def run_bang_command(
pass
return int(proc.returncode or 0)
hermes_cli/banner.py
+203 -321
View File
@@ -11,43 +11,19 @@ from urllib.parse import urlparse
from hermes_constants import get_hermes_home
from typing import TYPE_CHECKING, Any, Dict, List, Optional
# rich and prompt_toolkit are imported lazily (inside the functions that use
# them) rather than at module level. Importing this module is on the TUI
# gateway's critical startup path purely to reach the lightweight update-check
# helpers (``prefetch_update_check``); pulling rich.console + prompt_toolkit
# eagerly added ~50ms of wasted imports before ``gateway.ready`` could fire.
# Keep the type-only reference available to checkers without the runtime cost.
# 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
# =========================================================================
_GOLD = "\033[1;38;2;255;215;0m" # True-color #FFD700 bold
_BOLD = "\033[1m"
# ANSI building blocks for conversation display (``_DIM``/``_RST`` are imported by callbacks.py).
_DIM = "\033[2m"
_RST = "\033[0m"
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 or absent stdout (pythonw.exe,
# CI, `hermes ... > file`) raises NoConsoleScreenBufferError from its Win32Output — display
# helpers must never crash the caller over that, so degrade to plain print.
if _quiet(lambda: _pt_print(_PT_ANSI(text)) or True) is None:
print(text)
# =========================================================================
# Skin-aware color helpers
# =========================================================================
def _quiet(fn, default=None):
"""``fn()``, or ``default`` on any exception — for best-effort display inputs."""
try:
@@ -56,6 +32,16 @@ def _quiet(fn, default=None):
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, or None when the skin engine is unavailable."""
from hermes_cli.skin_engine import get_active_skin
@@ -65,9 +51,9 @@ def _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
# =========================================================================
# === ASCII Art & Branding ===
from hermes_cli import __version__ as VERSION, __release_date__ as RELEASE_DATE
@@ -94,10 +80,7 @@ HERMES_CADUCEUS = """[#CD7F32]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⡀⠀⣀⣀
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠳⠈⣡⠞⠁⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]"""
# =========================================================================
# Skills scanning
# =========================================================================
# === 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``).
@@ -105,7 +88,6 @@ _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
@@ -123,9 +105,8 @@ def _memo(cache_name: str, compute):
def get_available_skills() -> Dict[str, List[str]]:
"""Return skills grouped by category, filtered by platform and disabled state.
Cached per-process: this feeds only the startup banner, whose snapshot is taken once anyway, and
the underlying skills-tree walk costs ~100ms. ``prefetch_banner_data()`` uses the cache to pay
that walk off-thread. A failed scan yields ``{}`` and is not cached.
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
@@ -144,15 +125,11 @@ def get_available_skills() -> Dict[str, List[str]]:
return {} if result is _UNCACHED else result
# =========================================================================
# Update check
# =========================================================================
# === Update check ===
# Cache update check results for 6 hours to avoid repeated git fetches
_UPDATE_CHECK_CACHE_SECONDS = 6 * 3600
_UPDATE_CHECK_CACHE_SECONDS = 6 * 3600 # avoid repeated git fetches
# Sentinel returned when we know an update exists but can't count commits
# (e.g. nix-built hermes — no local git history to count against).
# Returned when an update is known to exist but commits can't be counted (e.g. nix builds).
UPDATE_AVAILABLE_NO_COUNT = -1
_UPSTREAM_REPO_URL = "https://github.com/NousResearch/hermes-agent.git"
@@ -189,25 +166,19 @@ def _git_run(
):
"""Run ``git <args>`` with the shared subprocess boilerplate; None on any exception.
git output is UTF-8; on Windows ``text=True`` defaults to the ANSI code page and bytes like 0x90
(3rd byte of 🐛 in a commit subject) crash the stdlib reader thread (#52649), hence the explicit
git output is UTF-8; on Windows ``text=True`` defaults to the ANSI code page and a byte like the
3rd of 🐛 in a commit subject crashes the stdlib reader thread (#52649), hence the explicit
encoding. ``network=True`` (ls-remote/fetch) detaches stdin and disables git/GCM prompts so a
passive update check can never hang on a ``Username for 'https://github.com':`` prompt.
"""
kwargs: dict = {}
if network:
from hermes_cli._subprocess_compat import noninteractive_git_env
kwargs = {"stdin": subprocess.DEVNULL, "env": noninteractive_git_env()}
try:
return subprocess.run(
["git", *args],
capture_output=True,
timeout=timeout,
cwd=str(cwd) if cwd is not None else None,
**(_GIT_TEXT_KW if text else {}),
**kwargs,
)
["git", *args], capture_output=True, timeout=timeout, cwd=str(cwd) if cwd is not None else None,
**(_GIT_TEXT_KW if text else {}), **kwargs)
except Exception:
return None
@@ -233,6 +204,10 @@ def _git_count(args: list[str], *, cwd: Path) -> Optional[int]:
return None
def _is_full_sha(value: Optional[str]) -> bool:
return isinstance(value, str) and len(value) == 40 and all(c in "0123456789abcdefABCDEF" for c in value)
def _github_compare_behind(current_rev: str, target_rev: str) -> Optional[int]:
"""Exact behind-count via the GitHub compare API for uncountable graphs.
@@ -241,17 +216,13 @@ def _github_compare_behind(current_rev: str, target_rev: str) -> Optional[int]:
"""
if not (_is_full_sha(current_rev) and _is_full_sha(target_rev)):
return None
url = (
"https://api.github.com/repos/nousresearch/hermes-agent/"
f"compare/{current_rev}...{target_rev}"
)
url = f"https://api.github.com/repos/nousresearch/hermes-agent/compare/{current_rev}...{target_rev}"
def _fetch():
import urllib.request
# api.github.com 403s requests without a User-Agent.
req = urllib.request.Request(
url, headers={"Accept": "application/vnd.github+json", "User-Agent": "hermes-cli-update-check"},
)
url, headers={"Accept": "application/vnd.github+json", "User-Agent": "hermes-cli-update-check"})
with urllib.request.urlopen(req, timeout=10) as resp:
return json.loads(resp.read().decode("utf-8"))
@@ -262,21 +233,13 @@ def _github_compare_behind(current_rev: str, target_rev: str) -> Optional[int]:
return None
def _behind_count_or_sentinel(local_rev: str, upstream_rev: str) -> int:
"""Exact behind-count via the compare API, else the honest no-count sentinel.
``ahead_by == 0`` with differing tips means the remote tip is reachable from our HEAD — a
local-ahead checkout, i.e. NOT behind. A local-only HEAD 404s on the API, which safely degrades
to ``UPDATE_AVAILABLE_NO_COUNT`` — never a fabricated 1.
"""
counted = _github_compare_behind(local_rev, upstream_rev)
return counted if counted is not None else UPDATE_AVAILABLE_NO_COUNT
def _tips_behind(head_rev: Optional[str], target_rev: Optional[str], repo_dir: Optional[Path] = None) -> Optional[int]:
"""Behind-count from two tip SHAs: None if either is unknown, 0 when equal, else count/sentinel.
With ``repo_dir``, a target that is already an ancestor of HEAD (local-ahead checkout) is 0 too.
``ahead_by == 0`` with differing tips means the remote tip is reachable from our HEAD — NOT
behind. A local-only HEAD 404s on the API, which degrades to ``UPDATE_AVAILABLE_NO_COUNT`` —
never a fabricated 1.
"""
if not head_rev or not target_rev:
return None
@@ -286,11 +249,8 @@ def _tips_behind(head_rev: Optional[str], target_rev: Optional[str], repo_dir: O
ancestor = _git_run(["merge-base", "--is-ancestor", target_rev, "HEAD"], cwd=repo_dir, text=False)
if ancestor is not None and ancestor.returncode == 0:
return 0
return _behind_count_or_sentinel(head_rev, target_rev)
def _is_full_sha(value: Optional[str]) -> bool:
return isinstance(value, str) and len(value) == 40 and all(c in "0123456789abcdefABCDEF" for c in value)
counted = _github_compare_behind(head_rev, target_rev)
return counted if counted is not None else UPDATE_AVAILABLE_NO_COUNT
def _upstream_main_sha() -> Optional[str]:
@@ -302,11 +262,7 @@ def _upstream_main_sha() -> Optional[str]:
def _check_via_rev(local_rev: str) -> Optional[int]:
"""Compare an embedded git revision to upstream main via ls-remote.
Returns 0 if up-to-date, the exact behind-count when the GitHub compare API can recover it,
``UPDATE_AVAILABLE_NO_COUNT`` if behind by an unknown amount, or ``None`` on failure.
"""
"""Compare an embedded git revision to upstream main via ls-remote (see ``_tips_behind``)."""
return _tips_behind(local_rev, _upstream_main_sha())
@@ -318,63 +274,49 @@ def _check_via_local_git(repo_dir: Path) -> Optional[int]:
if not head_rev:
return None
# Passive probe via HTTPS ls-remote (never SSH — no hardware-key prompts). Tip SHAs alone
# can't distinguish "behind" from a local carried commit sitting AHEAD of origin/main, and
# misreporting an ahead checkout as behind nudges the user into `hermes update`, which can
# wipe their carried work — hence the ancestor check, against the FRESH upstream SHA (not
# the possibly stale origin/main tracking ref) so a stale ref can't fake an up-to-date report.
# can't distinguish "behind" from a local commit AHEAD of origin/main, and misreporting an
# ahead checkout nudges the user into `hermes update`, which can wipe carried work — hence
# the ancestor check, against the FRESH upstream SHA (a stale tracking ref can't fake an
# up-to-date report).
return _tips_behind(head_rev, _upstream_main_sha(), repo_dir)
# Installer checkouts are shallow (`git clone --depth 1`). On a shallow
# clone the history stops at a single commit, so a plain `git fetch` would
# unshallow the repo (dragging in the whole history) and
# `rev-list --count HEAD..origin/main` would report a huge bogus "behind"
# number (e.g. "12492 commits behind"). Detect shallow up front: fetch with
# --depth 1 to preserve the boundary and compare tip SHAs instead of
# counting. Full clones (developers, Docker dev images) keep the exact
# count path unchanged. Mirrors the desktop fix in apps/desktop/electron/main.cjs.
# Installer checkouts are shallow (`git clone --depth 1`): a plain `git fetch` would unshallow
# the repo and `rev-list --count HEAD..origin/main` would report a bogus "12492 commits
# behind". Fetch with --depth 1 to preserve the boundary and compare tip SHAs instead. Full
# clones keep the exact count path. Mirrors apps/desktop/electron/main.cjs.
is_shallow = _git_stdout(["rev-parse", "--is-shallow-repository"], cwd=repo_dir) == "true"
def _fetch() -> bool:
# Self-heal abandoned git lock files before fetching. A stale .git/shallow.lock from a
# crashed fetch makes the fetch fail, the exception is swallowed, and stale refs get
# compared against HEAD — silently degrading the passive check until a human removes the
# lock (git never self-heals these). The passive check is also the main tmp_pack GENERATOR
# on flaky lines (several aborted fetches per day), so it must be the janitor too, or debris
# accumulates unbounded between manual updates (#93732).
# Self-heal abandoned git lock files first. A stale .git/shallow.lock from a crashed fetch
# makes every fetch fail silently and stale refs get compared against HEAD until a human
# removes the lock. This passive check is also the main tmp_pack GENERATOR on flaky lines,
# so it must be the janitor too (#93732).
from hermes_cli.gitlock import clear_stale_git_locks, clear_stale_tmp_packs
clear_stale_git_locks(repo_dir)
clear_stale_tmp_packs(repo_dir)
# Scope the fetch to the one branch the behind-count compares against. An unscoped
# ``git fetch origin`` transfers every remote head (~1,400 on this repo — measured 3.0 s vs
# 0.55 s scoped) and can burn the full 10 s timeout on slow links. Modern git updates the
# ``origin/main`` tracking ref on a scoped fetch, so the ``HEAD..origin/main`` count below
# is unaffected; the shallow path compares against FETCH_HEAD, which a scoped fetch also
# updates. ``--depth 1`` preserves the shallow boundary.
# Scope the fetch to the one branch compared against: an unscoped ``git fetch origin``
# transfers ~1,400 remote heads (3.0 s vs 0.55 s measured) and can burn the full timeout.
# A scoped fetch still updates ``origin/main`` and FETCH_HEAD; ``--depth 1`` preserves
# the shallow boundary.
fetch_args = ["fetch", "origin", "main", *(["--depth", "1"] if is_shallow else []), "--quiet"]
fetch_proc = _git_run(fetch_args, cwd=repo_dir, timeout=10, text=False, network=True)
return fetch_proc is not None and fetch_proc.returncode == 0
fetch_ok = _quiet(_fetch, False) # Offline or timeout — don't use stale refs
# When the fetch fails, the local origin/main tracking ref is stale. It
# cannot prove *currentness* (a 0 behind-count may just mean the stale ref
# hasn't caught up), but if it already shows HEAD behind, that is sound
# evidence an update exists — the ref was good at some point in the past.
# Return the positive stale count; return None (inconclusive) otherwise so
# the caller doesn't cache a false "up to date". (#82166, review #92578)
# When the fetch fails the local origin/main ref is stale: it cannot prove *currentness*, but
# if it already shows HEAD behind, that is sound evidence an update exists. Return the positive
# stale count; None (inconclusive) otherwise so the caller doesn't cache a false "up to date".
if is_shallow:
if not fetch_ok:
return None
# No history to count across the shallow boundary. `origin/main` may not be a tracking ref
# in a `clone --depth 1`, so prefer FETCH_HEAD (just updated by the fetch above) and fall
# back to origin/main. Tips differ but the shallow boundary hides the history between them.
# No history across the shallow boundary. `origin/main` may not be a tracking ref in a
# `clone --depth 1`, so prefer FETCH_HEAD (just updated) and fall back to origin/main.
head_rev = _git_stdout(["rev-parse", "HEAD"], cwd=repo_dir)
target_rev = (
_git_stdout(["rev-parse", "FETCH_HEAD"], cwd=repo_dir)
or _git_stdout(["rev-parse", "origin/main"], cwd=repo_dir)
)
or _git_stdout(["rev-parse", "origin/main"], cwd=repo_dir))
return _tips_behind(head_rev, target_rev)
behind = _git_count(["rev-list", "--count", "HEAD..origin/main"], cwd=repo_dir)
@@ -392,21 +334,16 @@ def _read_json(path: Path) -> Optional[dict]:
def check_for_updates() -> Optional[int]:
"""Check whether a Hermes update is available.
Two paths: if ``HERMES_REVISION`` is set (nix builds embed it), compare it to upstream main via
``git ls-remote``. Otherwise look for a local git checkout and count commits behind
``origin/main``.
If ``HERMES_REVISION`` is set (nix builds embed it), compare it to upstream main via
``git ls-remote``; otherwise count commits behind ``origin/main`` in the local checkout.
"""
hermes_home = get_hermes_home()
cache_file = hermes_home / ".update_check"
embedded_rev = os.environ.get("HERMES_REVISION") or None
# Docker images have no working tree to count commits against — the
# published image excludes `.git` (see .dockerignore) and sets no
# HERMES_REVISION (that's nix-only). Returning None makes both the Rich
# banner (build_welcome_banner) and the Ink badge (branding.tsx, guarded
# on `typeof === 'number' && > 0`) show nothing. The dashboard's REST
# `/api/hermes/update/check` endpoint short-circuits docker the same way
# (web_server.py); mirror that here so the banner/TUI surfaces agree.
# Docker images have no working tree (the image excludes `.git`) and set no HERMES_REVISION.
# None makes both the Rich banner and the Ink badge show nothing, mirroring the dashboard's
# `/api/hermes/update/check` short-circuit so the surfaces agree.
def _install_method():
from hermes_cli.config import detect_install_method, get_project_root
return detect_install_method(get_project_root())
@@ -414,60 +351,49 @@ def check_for_updates() -> Optional[int]:
if _quiet(_install_method) in {"docker", "apt"}:
return None
# Read cache — invalidate if the embedded rev OR installed version has
# changed since the last check.
# Cache is invalidated when the embedded rev OR installed version changed since the last check.
now = time.time()
cached = _read_json(cache_file)
if (
cached is not None
and now - cached.get("ts", 0) < _UPDATE_CHECK_CACHE_SECONDS
and cached.get("rev") == embedded_rev
and cached.get("ver") == VERSION
):
and cached.get("ver") == VERSION):
return cached.get("behind")
if embedded_rev:
behind = _check_via_rev(embedded_rev)
else:
# No git checkout and no embedded revision — can't determine update
# status (the Docker path, already short-circuited above, or an
# unsupported install without a source tree).
# No checkout and no embedded revision — status can't be determined.
repo_dir = _resolve_repo_dir()
behind = _check_via_local_git(repo_dir) if repo_dir is not None else None
# Don't cache inconclusive results (None). A None means the check could not run — typically a
# failed git fetch. Caching None would suppress retries for the full 6-hour cache window,
# leaving the user with a stale "up to date" or no information for hours after connectivity
# is restored (#82166).
# Don't cache inconclusive results: None means the check could not run (typically a failed
# fetch), and caching it would suppress retries for the full 6-hour window (#82166).
if behind is not None:
_quiet(lambda: cache_file.write_text(
json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION}),
encoding="utf-8",
))
json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION}), encoding="utf-8"))
return behind
def _resolve_repo_dir() -> Optional[Path]:
"""Return the active Hermes git checkout, or None if this isn't a git install.
"""The active Hermes git checkout, or None if this isn't a git install.
Prefers the running code's location over the profile-scoped path because ``$HERMES_HOME/hermes-
agent/`` may be a stale copy carried over by ``--clone-all``.
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():
hermes_home = get_hermes_home()
repo_dir = hermes_home / "hermes-agent"
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): the state costs 2-3 git subprocesses (~100ms)
and the checkout revision cannot change under a running CLI in a way the banner needs to observe
live. The cache also lets ``prefetch_banner_data()`` pay this cost off-thread before the banner
renders.
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)
@@ -502,11 +428,9 @@ _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 git tag, or None.
"""Return ``(tag, release_url)`` for the latest local git tag, or None (a miss is cached too).
Local-only — runs ``git describe --tags --abbrev=0`` against the Hermes checkout. Cached per-
process (a miss is cached too). Release URL always points at the canonical
NousResearch/hermes-agent repo (forks don't get a link).
Release URL always points at the canonical NousResearch/hermes-agent repo (forks get no link).
"""
def _compute():
rd = repo_dir or _resolve_repo_dir()
@@ -530,9 +454,7 @@ def format_banner_version_label() -> str:
return f"{base} · upstream {upstream} · local {local} (+{ahead} carried {_plural(ahead, 'commit')})"
# =========================================================================
# Non-blocking update check
# =========================================================================
# === Non-blocking update check ===
_update_result: Optional[int] = None
_update_check_done = threading.Event()
@@ -558,10 +480,9 @@ _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 of subprocesses) and the skills index (~110ms rglob) are cached per-process
by their own modules, so warming them while the main thread pays the CPU-bound ``cli`` /
prompt_toolkit imports overlaps GIL-releasing I/O with import work. Idempotent; failures
don't matter because the banner recomputes anything missing.
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:
@@ -587,11 +508,9 @@ def _format_update_notice(behind: int) -> str:
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-built hermes; we know an update
# exists but not by how much, and we don't know how the user
# installed it (nix run, profile, system flake, home-manager).
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}"
@@ -601,10 +520,10 @@ _deferred_update_notice_started = False
def _defer_update_notice(console: "Console", max_wait: float = 30.0) -> None:
"""Print the update warning once the prefetched check completes.
"""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. Prints at most once per process.
git/network.
"""
global _deferred_update_notice_started
if _deferred_update_notice_started:
@@ -618,9 +537,7 @@ def _defer_update_notice(console: "Console", max_wait: float = 30.0) -> None:
_daemon("update-notice", _wait_and_print) # never break the session over an update notice
# =========================================================================
# Welcome banner
# =========================================================================
# === Welcome banner ===
def _plural(n: int, word: str) -> str:
return word if n == 1 else f"{word}s"
@@ -648,19 +565,12 @@ def _short_label(name: str) -> str:
return name[:25] + "..." if len(name) > 28 else name
# =========================================================================
# Banner snapshot — warm-launch fast path
# =========================================================================
# The banner's tool panel needs the full tool registry (get_tool_definitions:
# tools/*.py discovery + every check_fn), which costs ~0.5-0.9s cold and is
# the single largest chunk of CLI time-to-banner. The tool list shown in the
# banner is a pure function of (config.yaml, .env, code checkout, enabled
# toolsets), so we snapshot the rendered inputs to disk after each launch
# and replay them on the next one when the fingerprint matches. The agent's
# REAL tool list is still computed fresh at first message (agent init) —
# the snapshot only feeds the cosmetic startup panel, and a background
# refresh re-verifies it right after the banner renders (see
# cli.show_banner), so a stale panel self-heals within one launch.
# === 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
@@ -672,7 +582,6 @@ def _banner_snapshot_path() -> Path:
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")
@@ -703,17 +612,13 @@ def load_banner_snapshot(enabled_toolsets: List[str] = None) -> Optional[Dict[st
if blob.get("enabled_toolsets") != sorted(enabled_toolsets or []):
return None
if not isinstance(blob.get("tools"), list) or not all(
isinstance(blob.get(k), dict) for k in ("toolset_map", "availability", "skills_by_category")
):
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],
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()
@@ -725,8 +630,7 @@ def save_banner_snapshot(
"tools": [
{"function": {"name": t["function"]["name"]}}
for t in tools
if isinstance(t, dict) and t.get("function", {}).get("name")
],
if isinstance(t, dict) and t.get("function", {}).get("name")],
"toolset_map": toolset_map,
"availability": {
"unavailable_toolsets": availability.get("unavailable_toolsets", []),
@@ -734,6 +638,7 @@ def save_banner_snapshot(
},
"skills_by_category": get_available_skills(),
}
def _write():
import tempfile
path = _banner_snapshot_path()
@@ -747,29 +652,23 @@ def save_banner_snapshot(
def compute_toolset_availability(enabled_toolsets: List[str] = None) -> Dict[str, Any]:
"""Compute the banner's toolset-availability payload.
"""Compute ``{"unavailable_toolsets", "lazy_tools", "disabled_tools"}`` for the banner.
Returns ``{"unavailable_toolsets", "lazy_tools", "disabled_tools"}``. Split out so the result
can be snapshotted and replayed on the next launch without importing ``model_tools``.
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 toolset registry, so it includes
# toolsets that aren't part of this agent's platform set at all (e.g.
# `discord`, `feishu_doc` on a CLI session). Those must never surface in the
# banner's "Available Tools" — they aren't exposed to the agent. Restrict to
# toolsets actually enabled for this agent; a toolset that's enabled but
# currently has unmet deps legitimately shows as disabled/lazy below.
# 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
]
# Tools whose toolset has a check_fn are lazy-initialized (e.g. honcho, homeassistant) — they
# show as unavailable at banner time because the check hasn't run yet, but aren't misconfigured.
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")
@@ -833,7 +732,6 @@ 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]
@@ -874,95 +772,50 @@ def _active_profile_name() -> Optional[str]:
return get_active_profile_name()
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):
"""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)
unavailable_toolsets = availability.get("unavailable_toolsets", [])
lazy_tools = set(availability.get("lazy_tools", []))
disabled_tools = set(availability.get("disabled_tools", []))
_enabled_ts = {str(t) for t in enabled_toolsets}
layout_table = Table.grid(padding=(0, 2))
layout_table.add_column("left", justify="center")
layout_table.add_column("right", justify="left")
# 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, ""]
def _banner_left_lines(model: str, cwd: str, session_id, context_length, provider, *, accent: str, dim: str) -> list:
"""Model / cwd / session lines under the hero art."""
def _dim_sep(label: str) -> str:
return f" [dim {dim}]·[/] [dim {dim}]{label}[/]"
lines = []
ctx_str = _dim_sep(f"{_format_context_length(context_length)} context") if context_length else ""
nous_str = _dim_sep("Nous Research")
if (provider or "").strip().lower() == "moa":
# MoA virtual provider: ``model`` is a preset name. Show the preset and
# its aggregator so the banner is meaningful instead of a bare slug.
# 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 ""
left_lines.append(f"[{accent}]MoA: {_short_label(model)}[/]{agg_str}{ctx_str}{nous_str}")
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: say so in red instead of a blank/"unknown"
# slug — this is the single clearest place to tell the user what
# is wrong and how to fix it.
left_lines.append(
f"[bold red]no model configured[/] "
f"[dim {dim}]— run /model or hermes setup[/]"
)
# 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")
left_lines.append(f"[{accent}]{_short_label(model_short)}[/]{ctx_str}{nous_str}")
lines.append(f"[{accent}]{_short_label(model_short)}[/]{ctx_str}{nous_str}")
if os.getenv("HERMES_YOLO_MODE"):
left_lines.append(f"[bold red]⚠ YOLO mode[/] [dim {dim}]— all approval prompts bypassed[/]")
left_lines.append(f"[dim {dim}]{cwd}[/]")
lines.append(f"[bold red]⚠ YOLO mode[/] [dim {dim}]— all approval prompts bypassed[/]")
lines.append(f"[dim {dim}]{cwd}[/]")
if session_id:
left_lines.append(f"[dim {_skin_color('session_border', '#8B8682')}]Session: {session_id}[/]")
left_content = "\n".join(left_lines)
lines.append(f"[dim {_skin_color('session_border', '#8B8682')}]Session: {session_id}[/]")
return lines
right_lines = [f"[bold {accent}]Available Tools[/]"]
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)
sorted_toolsets = sorted(toolsets_dict.keys())
def _color_tool(name: Optional[str]) -> str:
if name is None: # truncation marker
return "[dim]...[/]"
@@ -972,50 +825,84 @@ def build_welcome_banner(console: "Console", model: str, cwd: str,
return f"[yellow]{name}[/]"
return f"[{text}]{name}[/]"
sorted_toolsets = sorted(toolsets_dict.keys())
for toolset in sorted_toolsets[:8]:
tool_names = _truncate_tool_names(sorted(toolsets_dict[toolset]))
right_lines.append(f"[dim {dim}]{toolset}:[/] {', '.join(_color_tool(n) for n in tool_names)}")
lines.append(f"[dim {dim}]{toolset}:[/] {', '.join(_color_tool(n) for n in tool_names)}")
if len(sorted_toolsets) > 8:
right_lines.append(f"[dim {dim}](and {len(sorted_toolsets) - 8} more toolsets...)[/]")
lines.append(f"[dim {dim}](and {len(sorted_toolsets) - 8} more toolsets...)[/]")
return lines
# MCP Servers section (only if configured). Probe cheaply first: the
# full get_mcp_status() path resolves portable plugin MCP servers,
# which JOINS the in-flight background plugin discovery (~100ms on the
# startup path). When neither config.yaml nor the persisted plugin
# key cache mentions any MCP server, skip the section outright.
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,
):
"""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)
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
# (it exposes skill_view / skill_manage). When it's disabled — e.g. a Blank
# Slate install — the agent literally cannot load any skill, so advertising
# the on-disk catalog here is misleading. Reflect the real state instead.
# 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())
# Dynamically size skills display based on terminal width.
# Rich grid with 2 columns; right column gets roughly 60% of terminal.
_term_cols = shutil.get_terminal_size().columns
_right_col_width = max(int(_term_cols * 0.6) - 10, 30)
if not _skills_enabled:
right_lines.append(f"[dim {dim}]Skills toolset disabled[/]")
elif skills_by_category:
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))
right_lines.append(f"[dim {dim}]{category}:[/] [{text}]{skills_str}[/]")
else:
right_lines.append(f"[dim {dim}]No skills installed[/]")
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"])
@@ -1023,14 +910,12 @@ def build_welcome_banner(console: "Console", model: str, cwd: str,
if mcp_connected:
summary_parts.append(f"{mcp_connected} MCP servers")
summary_parts.append("/help for commands")
# Indicate when the codex_app_server runtime is active so users
# understand why tool counts may not match what's actually reachable
# (codex builds its own tool list inside the spawned subprocess).
# 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)[/]"
)
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":
@@ -1038,13 +923,10 @@ def build_welcome_banner(console: "Console", model: str, cwd: str,
right_lines.append(f"[dim {dim}]{' · '.join(summary_parts)}[/]")
# Update check — use prefetched result if available. NEVER block the
# banner on it: the prefetch does git/network work that rarely finishes
# before the banner renders, so a blocking wait here just adds its full
# timeout to every startup (500ms of the banner path pre-fix). If the
# result isn't ready yet, defer the warning line: a daemon thread waits
# for the prefetch and prints the same notice above the prompt when it
# lands (prompt_toolkit's patch_stdout renders late prints safely).
# 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
# (prompt_toolkit's patch_stdout renders late prints safely).
def _update_line():
behind = get_update_result(timeout=0.05)
if behind is None and not _update_check_done.is_set():
@@ -1054,18 +936,18 @@ def build_welcome_banner(console: "Console", model: str, cwd: str,
_quiet(_update_line) # Never break the banner over an update check
layout_table.add_row(left_content, "\n".join(right_lines))
layout_table = Table.grid(padding=(0, 2))
layout_table.add_column("left", justify="center")
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),
)
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: