"""Informational views and reload flows for the interactive CLI: banner, help, tools, usage, insights, MCP/skills reload, bang shell. Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. cli.py-internal symbols are imported LAZILY inside each method — the mixin never imports ``cli`` at module load time (cycle). """ from __future__ import annotations import concurrent.futures import logging import os import shutil import threading import time from agent.i18n import t from hermes_constants import is_termux as _is_termux_environment from rich.markup import escape as _escape from utils import base_url_hostname, file_signature from hermes_cli.cli_modal_mixin import _gated_confirm from hermes_cli.colors import Colors as _Colors CONFIG_WATCH_INTERVAL = 5.0 # seconds between config.yaml stat() calls _TOOL_PROGRESS_CYCLE = ["off", "new", "all", "verbose"] # Raw ANSI (not Rich markup): _cprint routes through prompt_toolkit's renderer, while Rich markup # written to stdout gets mangled by patch_stdout's StdoutProxy ('?[33mTool progress: NEW?[0m', #2262). _TOOL_PROGRESS_STYLES = { "off": _Colors.DIM, "new": _Colors.YELLOW, "all": _Colors.GREEN, "verbose": f"{_Colors.BOLD}{_Colors.GREEN}", } def _tool_progress_label(mode: str) -> str: """Localized ``Tool progress: MODE — detail`` line for /verbose ("" for an unknown mode).""" style = _TOOL_PROGRESS_STYLES.get(mode) if style is None: return "" return f"{style}{t(f'cli.verbose.label_{mode}')}{_Colors.RESET} — {t(f'cli.verbose.detail_{mode}')}" def _reload_mcp_choices() -> list[tuple[str, str, str]]: """(id, label, hint) rows for the /reload-mcp confirm modal, in the active language.""" return [ ("once", t("cli.reload_mcp.choice_once"), t("cli.reload_mcp.choice_once_hint")), ("always", t("cli.reload_mcp.choice_always"), t("cli.reload_mcp.choice_always_hint")), ("cancel", t("cli.reload_mcp.choice_cancel"), t("cli.reload_mcp.choice_cancel_hint"))] # /help section headers: the category ids are load-bearing for gateway help and the registry, so # they stay identifiers and are only localized at render time. _HELP_SECTION_KEYS = { "Session": "session", "Configuration": "configuration", "Tools & Skills": "tools_skills", "Info": "info", "Exit": "exit", "Context": "context", "Background & Automation": "background_automation", } def _help_section_title(category: str) -> str: slug = _HELP_SECTION_KEYS.get(category) return t(f"cli.help.section_{slug}") if slug else category def _ascii_box(title: str, width: int) -> None: """Print the kawaii ``+---+ | title | +---+`` header used by /tools and /toolsets.""" pad = width - len(title) print("+" + "-" * width + "+") print("|" + " " * (pad // 2) + title + " " * (pad - pad // 2) + "|") print("+" + "-" * width + "+") def _toolset_map(tools, availability, get_toolset_for_tool) -> dict: """tool name → toolset id, including tools of unavailable toolsets (banner snapshot).""" tmap = {tool["function"]["name"]: get_toolset_for_tool(tool["function"]["name"]) for tool in tools} for item in availability.get("unavailable_toolsets", []): for name in item.get("tools", []): tmap.setdefault(name, item.get("id", item.get("name", ""))) return tmap def _skill_line(item: dict) -> str: nm = item.get("name", "") desc = item.get("description", "") return f" - {nm}: {desc}" if desc else f" - {nm}" class CLIInfoMixin: """Informational views and reload flows for the interactive CLI: banner, help, tools, usage, insights, MCP/skills reload, bang shell.""" def show_banner(self): """Display the welcome banner in Claude Code style.""" from cli import _build_compact_banner, get_tool_definitions, logger from hermes_cli.banner import build_welcome_banner self.console.clear() ctx_len = None if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'): ctx_len = self.agent.context_compressor.context_length from agent.context_pin import is_context_pinned ctx_pinned = is_context_pinned(ctx_len, getattr(getattr(self, "agent", None), "_config_context_length", None)) # Auto-compact for narrow terminals — the full banner needs ~80 columns to avoid wrapping. if self.compact or shutil.get_terminal_size().columns < 80: self._console_print(_build_compact_banner()) self._show_status() else: # Warm-launch fast path: replay last launch's tool panel when the snapshot fingerprint # (config.yaml + .env + checkout rev + toolsets) is unchanged, skipping the ~0.5-0.9s # cold get_tool_definitions walk. The agent's REAL tool list is still computed fresh at # first message; a background refresh re-verifies the snapshot so drift self-heals. from hermes_cli.banner import ( compute_toolset_availability, load_banner_snapshot, save_banner_snapshot) try: snapshot = load_banner_snapshot(self.enabled_toolsets) except Exception: snapshot = None cwd = os.getenv("TERMINAL_CWD", os.getcwd()) # where commands will execute banner_kw = dict( console=self.console, model=self.model, cwd=cwd, enabled_toolsets=self.enabled_toolsets, session_id=self.session_id, context_length=ctx_len, provider=self.provider, context_pinned=ctx_pinned) if snapshot is not None: self._defer_tool_warnings = True toolset_map = snapshot["toolset_map"] build_welcome_banner( tools=snapshot["tools"], get_toolset_for_tool=lambda name: toolset_map.get(name), availability=snapshot["availability"], skills_by_category=snapshot.get("skills_by_category"), **banner_kw) def _refresh_banner_snapshot() -> None: try: from model_tools import get_toolset_for_tool tools = get_tool_definitions( enabled_toolsets=self.enabled_toolsets, disabled_toolsets=self.disabled_toolsets, quiet_mode=True) availability = compute_toolset_availability(self.enabled_toolsets) tmap = _toolset_map(tools, availability, get_toolset_for_tool) save_banner_snapshot(tools, self.enabled_toolsets, availability, tmap) except Exception: logger.debug("banner snapshot refresh failed", exc_info=True) threading.Thread( target=_refresh_banner_snapshot, name="banner-snapshot-refresh", daemon=True, ).start() else: # Cold path: compute live, then persist the snapshot for the next launch. from model_tools import get_toolset_for_tool tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, disabled_toolsets=self.disabled_toolsets, quiet_mode=True) availability = compute_toolset_availability(self.enabled_toolsets) build_welcome_banner(tools=tools, availability=availability, **banner_kw) try: tmap = _toolset_map(tools, availability, get_toolset_for_tool) save_banner_snapshot(tools, self.enabled_toolsets, availability, tmap) except Exception: logger.debug("banner snapshot save failed", exc_info=True) # Tool discovery is deferred on the Termux bare prompt path (warnings show once tools # init). On the snapshot fast path the check walks every check_fn (~180ms) — run it in # the background and let its output land above the prompt (patch_stdout-safe). if os.environ.get("HERMES_DEFER_AGENT_STARTUP") != "1": if getattr(self, "_defer_tool_warnings", False): threading.Thread( target=self._show_tool_availability_warnings, name="tool-availability-warnings", daemon=True).start() else: self._show_tool_availability_warnings() # Low context warning — tied to the runtime guard so guidance cannot drift. from agent.model_metadata import MINIMUM_CONTEXT_LENGTH, is_local_endpoint if ctx_len and ctx_len < MINIMUM_CONTEXT_LENGTH: self._console_print() self._console_print(f"[yellow]{t('cli.banner.context_too_low', tokens=f'{ctx_len:,}')}[/]") self._console_print(f"[dim] {t('cli.banner.context_minimum', tokens=f'{MINIMUM_CONTEXT_LENGTH:,}')}[/]") base_url = getattr(self, "base_url", "") or "" from urllib.parse import urlparse as _urlparse try: _port = _urlparse(base_url if "://" in base_url else f"//{base_url}").port except ValueError: _port = None if _port == 11434 or "ollama" in base_url_hostname(base_url): fix = t("cli.banner.fix_ollama", tokens=str(MINIMUM_CONTEXT_LENGTH)) elif _port == 1234: fix = t("cli.banner.fix_lm_studio") elif is_local_endpoint(base_url): # llama.cpp / vLLM / any local server — not Ollama fix = t("cli.banner.fix_local_server", tokens_k=str(MINIMUM_CONTEXT_LENGTH // 1000), tokens=str(MINIMUM_CONTEXT_LENGTH)) else: fix = t("cli.banner.fix_config") self._console_print(f"[dim] {fix}[/]") from hermes_cli.model_switch import is_nous_hermes_non_agentic if is_nous_hermes_non_agentic(getattr(self, "model", "") or ""): self._console_print() self._console_print(f"[bold yellow]{t('cli.banner.hermes_models_not_agentic')}[/]") self._console_print(f"[dim] {t('cli.banner.hermes_models_lack_tools')}[/]") self._console_print(f"[dim] {t('cli.banner.hermes_models_switch_hint')}[/]") # Project-local skills one-liner: trusted → count; untrusted-with-skills → point at # `hermes skills trust`. Never raises. try: from agent.skill_utils import ( get_project_skills_dirs, get_untrusted_project_skills_root, iter_skill_index_files) _proj_dirs = get_project_skills_dirs() if _proj_dirs: _n = sum(sum(1 for _ in iter_skill_index_files(d, "SKILL.md")) for d in _proj_dirs) if _n: self._console_print(f"[dim]{t('cli.banner.project_skills_loaded', count=str(_n))}[/]") else: _untrusted = get_untrusted_project_skills_root() if _untrusted is not None: _root, _n = _untrusted self._console_print( f"[yellow]{t('cli.banner.project_skills_untrusted', count=str(_n), root=str(_root))}[/]") except Exception: logger.debug("project skills banner notice failed", exc_info=True) self._console_print() def _fast_command_available(self) -> bool: try: from hermes_cli.models import model_supports_fast_mode except Exception: return False agent = getattr(self, "agent", None) return model_supports_fast_mode(getattr(agent, "model", None) or getattr(self, "model", None)) def _command_available(self, slash_command: str) -> bool: if slash_command == "/fast": return self._fast_command_available() return True def show_help(self, arg: str = ""): """Display help. Bare /help shows categorized core commands with the skill list collapsed to one line; /help skills lists all skill commands; /help filters by substring.""" from cli import ( ChatConsole, _BOLD, _DIM, _RST, _accent_hex, _cprint, _ensure_skill_commands, _termux_example_image_path, get_skill_bundles) from hermes_cli.commands import COMMAND_REGISTRY, HELP_SESSION_SUBGROUPS arg = (arg or "").strip() skill_commands = _ensure_skill_commands() def _row(cmd: str, desc: str, width: int = 15) -> None: ChatConsole().print( f" [bold {_accent_hex()}]{cmd:<{width}}[/] [dim]-[/] {_escape(desc)}") # /help skills — the full list, kept out of the default view so core commands don't # scroll off screen. if arg.lower() in ("skills", "skill"): from agent.skill_commands import skill_command_collision_note from tools.skills_tool import _find_all_skills if skill_commands: _cprint(f"\n ⚡ {_BOLD}{t('cli.help.skill_commands')}{_RST} " f"({t('cli.help.n_installed', count=str(len(skill_commands)))}):") for cmd, info in sorted(skill_commands.items()): _row(cmd, info['description'], 22) else: _cprint(f"\n {t('cli.help.no_skill_commands')}") # Skills whose name is a built-in command never get a / (agent.skill_commands guard). for note in filter(None, (skill_command_collision_note(s["name"]) for s in _find_all_skills())): _cprint(f" {_DIM}⚠ {note}{_RST}") _cprint("") return query = arg.lower() if arg else "" default_header = t("cli.help.header") try: from hermes_cli.skin_engine import get_active_help_header header = get_active_help_header(default_header) except Exception: header = default_header header = ((header or "").strip() or default_header)[:55] _cprint(f"\n{_BOLD}+{'-' * 55}+{_RST}") _cprint(f"{_BOLD}|{header:^55}|{_RST}") _cprint(f"{_BOLD}+{'-' * 55}+{_RST}") def _section(title: str, rows) -> None: """Print available/matching rows under a `── title ──` header (omitted if empty).""" printed_header = False for cmd, desc in rows: if not self._command_available(cmd): continue if query and query not in cmd.lower() and query not in desc.lower(): continue if not printed_header: _cprint(f"\n {_BOLD}── {_help_section_title(title)} ──{_RST}") printed_header = True _row(cmd, desc) # Rows come straight from the registry so descriptions resolve through the localized # ``CommandDef.describe()`` accessor (names/aliases stay identifiers). by_category: dict[str, list[tuple[str, str]]] = {} for cmd_def in COMMAND_REGISTRY: if cmd_def.gateway_only: continue desc = cmd_def.describe() rows = by_category.setdefault(cmd_def.category, []) rows.append((f"/{cmd_def.name}", t("cli.help.usage_suffix", description=desc, command=cmd_def.name, args=cmd_def.args_hint) if cmd_def.args_hint else desc)) for alias in cmd_def.aliases: rows.append((f"/{alias}", t("cli.help.alias_suffix", description=desc, command=cmd_def.name))) for category, commands in by_category.items(): if category != "Session": _section(category, commands) continue # The oversized Session category renders as sub-groups # (Session / Context / Background & Automation). sub_of = {f"/{n}": sub for sub, names in HELP_SESSION_SUBGROUPS.items() for n in names} buckets: dict[str, list[tuple[str, str]]] = {"Session": []} for _sub in HELP_SESSION_SUBGROUPS: buckets[_sub] = [] for cmd, desc in commands: buckets[sub_of.get(cmd, "Session")].append((cmd, desc)) for _sub in ("Session", *HELP_SESSION_SUBGROUPS.keys()): _section(_sub, buckets.get(_sub) or []) # Skill commands collapse to a one-line pointer by default so 60+ entries don't bury the # core reference; filter mode includes matching skill commands inline. if query: matched_skills = [ (cmd, info) for cmd, info in sorted(skill_commands.items()) if query in cmd.lower() or query in (info.get("description", "").lower())] if matched_skills: _cprint(f"\n ⚡ {_BOLD}{t('cli.help.skill_commands')}{_RST} " f"({t('cli.help.matching_query', query=arg)}):") for cmd, info in matched_skills: _row(cmd, info['description'], 22) elif skill_commands: _cprint( f"\n ⚡ {_BOLD}{t('cli.help.skill_commands')}{_RST}: " f"{t('cli.help.n_installed', count=str(len(skill_commands)))} " f"— {_DIM}/help skills{_RST} {t('cli.help.to_list_them')}") _bundles_now = get_skill_bundles() if _bundles_now and not query: _cprint(f"\n ▣ {_BOLD}{t('cli.help.skill_bundles')}{_RST} " f"({t('cli.help.n_installed', count=str(len(_bundles_now)))}):") for cmd, info in sorted(_bundles_now.items()): skill_count = len(info.get("skills", [])) desc = info.get("description") or t("cli.help.load_n_skills", count=str(skill_count)) ChatConsole().print( f" [bold {_accent_hex()}]{cmd:<22}[/] [dim]-[/] " f"{_escape(desc)} [dim]({t('cli.help.n_skills', count=str(skill_count))})[/]") quick_commands = self.config.get("quick_commands", {}) if quick_commands and not query: _cprint(f"\n ⚡ {_BOLD}{t('cli.help.quick_commands')}{_RST} " f"({t('cli.help.n_configured', count=str(len(quick_commands)))}):") for name, qcmd in sorted(quick_commands.items()): _row('/' + name, qcmd.get("description", qcmd.get("type", "")), 22) if query: _cprint(f"\n {_DIM}{t('cli.help.filtered_by', query=arg)}{_RST}\n") return _cprint(f"\n {_DIM}{t('cli.help.tip_line')}{_RST}") _cprint(f" {_DIM}{t('cli.help.multiline_hint')}{_RST}") _cprint(f" {_DIM}{t('cli.help.draft_editor_hint')}{_RST}") if _is_termux_environment(): _cprint(f" {_DIM}{t('cli.help.attach_image_hint', path=_termux_example_image_path())}{_RST}\n") else: _cprint(f" {_DIM}{t('cli.help.paste_image_hint')}{_RST}\n") def show_tools(self): """Display available tools with kawaii ASCII art.""" from cli import get_tool_definitions from model_tools import get_toolset_for_tool # Pre-assembly list: /tools is a discovery surface, so it must show the full catalog # including tools deferred behind the tool_search bridge (users verify MCP installs here). tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, disabled_toolsets=self.disabled_toolsets, quiet_mode=True, skip_tool_search_assembly=True) if not tools: print(t("cli.tools.none_available")) return print() _ascii_box(t("cli.tools.header"), 78) print() toolsets: dict[str, list] = {} for tool in sorted(tools, key=lambda item: item["function"]["name"]): name = tool["function"]["name"] toolset = get_toolset_for_tool(name) or "unknown" desc = tool["function"].get("description", "").split("\n")[0] # First sentence: split on ". " (period+space) so "e.g." / "v2.0" stay intact. if ". " in desc: desc = desc[:desc.index(". ") + 1] toolsets.setdefault(toolset, []).append((name, desc)) for toolset in sorted(toolsets.keys()): print(f" [{toolset}]") for name, desc in toolsets[toolset]: print(f" * {name:<20} - {desc}") print() print(f" {t('cli.tools.total', count=str(len(tools)))}") print() def show_toolsets(self): """Display available toolsets with kawaii ASCII art.""" from toolsets import get_all_toolsets, get_toolset_info all_toolsets = get_all_toolsets() print() _ascii_box(t("cli.toolsets.header"), 58) print() for name in sorted(all_toolsets.keys()): info = get_toolset_info(name) if info: marker = "(*)" if self.enabled_toolsets and name in self.enabled_toolsets else " " print(f" {marker} {name:<18} [{info['tool_count']:>2} tools] - {info['description']}") print() print(f" {t('cli.toolsets.currently_enabled_legend')}") print() print(f" {t('cli.toolsets.tip_enable_all')}") print(f" {t('cli.toolsets.example')}") print() def _handle_whoami_command(self): """Display slash-command access for the local CLI surface.""" import getpass try: user_name = getpass.getuser() or "?" except Exception: user_name = "?" print() print(f" {t('cli.whoami.you_label'):<15} {t('cli.whoami.you_value')}") print(f" {t('cli.whoami.user_label'):<15} {user_name}") print(f" {t('cli.whoami.tier_label'):<15} {t('cli.whoami.tier_value')}") print(f" {t('cli.whoami.slash_commands_label'):<15} {t('cli.whoami.slash_commands_value')}") print() def _busy_inline_command(self, text: str, has_images: bool, names: tuple) -> bool: """True when ``text`` is a slash command in ``names`` typed while the agent is running. Such commands MUST bypass the normal ``_pending_input`` → ``process_loop`` path: the loop is blocked inside ``self.chat()`` for the whole run, so by the time the queued command is pulled, ``_agent_running`` has flipped back to False and it would be delivered as a next-turn message. Dispatching inline on the UI thread acts mid-run (``agent.steer()`` is thread-safe; ``/bg`` / ``/btw`` start their side session without touching the foreground turn). """ from cli import _looks_like_slash_command if not text or has_images or not _looks_like_slash_command(text): return False if not getattr(self, "_agent_running", False): return False try: from hermes_cli.commands import resolve_command cmd = resolve_command(text.split(None, 1)[0].lower().lstrip('/')) return bool(cmd and cmd.name in names) except Exception: return False def _should_handle_steer_command_inline(self, text: str, has_images: bool = False) -> bool: """Return True when /steer or /queue should be dispatched immediately while the agent is running. Queued raw, ``/queue `` only re-enqueued itself after the turn and ``/queue list|rm|edit`` could not inspect the queue until it had already drained.""" return self._busy_inline_command(text, has_images, ("steer", "queue")) def _should_handle_background_command_inline( self, text: str, has_images: bool = False) -> bool: """Return True when /bg or /btw should be dispatched while the agent runs (their ``CommandDef`` entries declare ``busy_policy="dispatch"``; the classic CLI honours it here).""" return self._busy_inline_command(text, has_images, ("bg", "btw")) def handle_bang_shell(self, text: str) -> bool: """Run a ``!`` submission. Returns True when it was handled. Dispatched from the input loop BEFORE slash routing and before anything is queued for the agent, so a bang command never becomes a turn: nothing touches ``conversation_history``, zero tokens, role alternation / prompt caching untouched by construction (tests/hermes_cli/test_bang_shell_mode.py). Returns False when the text is not a bang command or bang mode is disabled for this context (gateway/cron), so the caller routes normally. """ from cli import _rich_text_from_ansi from hermes_cli.bang_shell import ( USAGE_HINT, bang_shell_enabled, check_bang_approval, is_bang_command, parse_bang_command, resolve_bang_cwd, run_bang_command) if not is_bang_command(text): return False if not bang_shell_enabled(): # Gateway / cron / API: no composer, no human at a keyboard, and those users already # have shells — route normally rather than becoming remote execution. return False command = parse_bang_command(text) if not command: # bare `!` — show what the feature does self._console_print(f"[dim]{USAGE_HINT}[/]") return True approval = check_bang_approval(command) if not approval.get("approved"): message = approval.get("message") or t( "cli.shell.command_denied", reason=approval.get("description") or t("cli.shell.flagged_as_dangerous")) self._console_print(f"[bold red]{_escape(str(message))}[/]") return True exit_code = run_bang_command( command, cwd=resolve_bang_cwd(getattr(self, "session_id", None)), writer=lambda line: self._console_print(_rich_text_from_ansi(line))) if exit_code: self._console_print(f"[dim]{t('cli.shell.exited', code=str(exit_code))}[/]") return True def _show_gateway_status(self): """Show status of the gateway and connected messaging platforms.""" from hermes_constants import display_hermes_home from gateway.config import load_gateway_config, Platform print() _ascii_box(t("cli.gateway_status.header"), 60) print() try: config = load_gateway_config() print(f" {t('cli.gateway_status.platform_config_header')}") print(" " + "-" * 55) platform_status = { Platform.TELEGRAM: ("Telegram", "TELEGRAM_BOT_TOKEN"), Platform.DISCORD: ("Discord", "DISCORD_BOT_TOKEN"), Platform.SLACK: ("Slack", "SLACK_BOT_TOKEN"), Platform.WHATSAPP: ("WhatsApp", "WHATSAPP_ENABLED")} for platform, (name, env_var) in platform_status.items(): pconfig = config.platforms.get(platform) if pconfig and pconfig.enabled: home = config.get_home_channel(platform) home_str = f" → {home.name}" if home else "" print(f" ✓ {name:<12} {t('cli.gateway_status.enabled')}{home_str}") else: print(f" ○ {name:<12} {t('cli.gateway_status.not_configured', env_var=env_var)}") print() print(f" {t('cli.gateway_status.conversations_persist')}") print() print(f" {t('cli.gateway_status.to_start')}") print(" python cli.py --gateway") print() print(f" {t('cli.gateway_status.config_file', path=f'{display_hermes_home()}/config.yaml')}") print() except Exception as e: print(f" {t('cli.gateway_status.load_error', error=str(e))}") print() print(f" {t('cli.gateway_status.to_configure')}") print(f" {t('cli.gateway_status.step_env_vars')}") print(" TELEGRAM_BOT_TOKEN=your_token") print(" DISCORD_BOT_TOKEN=your_token") print(f" {t('cli.gateway_status.step_config', path=f'{display_hermes_home()}/config.yaml')}") print() def _print_random_tip(self) -> None: """Best-effort discovery tip (startup + /clear); never raises.""" try: from hermes_cli.tips import get_random_tip _tip = get_random_tip() try: from hermes_cli.skin_engine import get_active_skin _tip_color = get_active_skin().get_color("banner_dim", "#B8860B") except Exception: _tip_color = "#B8860B" self._console_print(f"[dim {_tip_color}]{t('cli.tip_line', tip=_tip)}[/]") except Exception: pass def _toggle_verbose(self): """Cycle tool progress mode: off → new → all → verbose → off. Tool-progress display is INDEPENDENT of global DEBUG logging: this never changes ``self.verbose`` or the agent's ``verbose_logging`` / ``quiet_mode`` (those belong to ``-v`` and ``/verbose-logging``). """ from cli import _cprint, save_config_value try: idx = _TOOL_PROGRESS_CYCLE.index(self.tool_progress_mode) except ValueError: idx = 2 # default to "all" self.tool_progress_mode = _TOOL_PROGRESS_CYCLE[(idx + 1) % len(_TOOL_PROGRESS_CYCLE)] # /verbose is the explicit tool-progress control, so cycling it takes ownership of the # mode back from focus view (else a "focus" badge + hidden-line counts would show while # tool lines visibly print). Display-only state change. if getattr(self, "_focus_view_enabled", False): self._focus_view_enabled = False self._focus_saved_tool_progress = None self._focus_hidden_lines = 0 self._focus_last_counted_tool = None try: from hermes_cli.focus_view import FOCUS_CONFIG_KEY save_config_value(FOCUS_CONFIG_KEY, False) except Exception: pass if self.agent: self.agent.reasoning_callback = self._current_reasoning_callback() # Sync the live agent so tool_executor rendering reflects the new mode this turn. self.agent.tool_progress_mode = self.tool_progress_mode _cprint(_tool_progress_label(self.tool_progress_mode)) def _handle_usage_command(self, cmd_original: str): """Dispatch `/usage [reset [--force]]`: bare `/usage` is the classic display; `reset` redeems one banked Codex rate-limit reset credit (refuses unless exhausted or --force).""" parts = cmd_original.split() args = [p.lower() for p in parts[1:]] if args and args[0] == "reset": self._usage_reset(force="--force" in args[1:]) return if args: print(f" {t('cli.usage.unknown_subcommand', args=' '.join(parts[1:]))}") return self._show_usage() def _agent_or_self(self, name: str): """Provider-ish attribute from the live agent, falling back to the CLI's own value.""" return (getattr(self.agent, name, None) if self.agent else None) or getattr(self, name, None) def _usage_reset(self, force: bool = False): """`/usage reset [--force]` — redeem one banked Codex reset credit.""" if str(self._agent_or_self("provider") or "").strip().lower() != "openai-codex": print(f" {t('cli.usage.reset_wrong_provider')}") print(f" {t('cli.usage.reset_switch_hint')}") return from agent.account_usage import redeem_codex_reset_credit print(f" {t('cli.usage.checking_reset_credits')}") with concurrent.futures.ThreadPoolExecutor(max_workers=1) as _pool: try: result = _pool.submit( redeem_codex_reset_credit, base_url=self._agent_or_self("base_url"), api_key=self._agent_or_self("api_key"), force=force, ).result(timeout=45.0) except concurrent.futures.TimeoutError: print(f" {t('cli.usage.codex_timeout')}") return print(f" {result.message}") def _show_context_breakdown(self, cmd_original: str = ""): """`/context [all]` — 5×20 glyph grid (cell ≈ 1% of the window) plus an estimated per-category table; `all` appends per-skill / per-toolset costs. Read-only: same chars/4 engine as the desktop popover (agent.context_breakdown) — no provider calls, no cache impact.""" if not self.agent: print(f" {t('cli.shared.no_active_agent')}") return args = cmd_original.split(maxsplit=1)[1].strip().lower() if " " in cmd_original else "" expanded = args in {"all", "full", "details"} from agent.context_breakdown import ( compute_context_details, compute_session_context_breakdown, render_context_breakdown_lines) try: payload = compute_session_context_breakdown(self.agent, self.conversation_history) except Exception as e: print(f" {t('cli.context.compute_failed', error=str(e))}") return details = None if expanded: try: details = compute_context_details(self.agent) except Exception: details = {"skills": [], "toolsets": []} from agent.context_file_sources import context_file_sources_for_agent, render_context_file_lines try: file_lines = render_context_file_lines(context_file_sources_for_agent(self.agent)) except Exception: file_lines = [] print() print(f" {t('cli.context.header', model=payload.get('model') or self.model)}") print() for line in render_context_breakdown_lines(payload, details=details, grid=True) + ([""] + file_lines if file_lines else []): print(f" {line}") print() def _show_usage(self): """Rate limits + session token usage (when a live agent exists) + Nous credits. The Nous credits block is agent-independent (portal fetch), so it runs even with no live agent — the TUI's /usage slash-worker resumes the session WITHOUT building an agent. """ from cli import datetime, format_duration_compact def _credits_or(fallback: str) -> None: # Account limits (e.g. Codex subscription windows) need only the configured provider # plus on-disk credentials, so they render without a live agent too (#42904). shown = self._print_account_limits() if self._print_nous_credits_block(): self._print_usage_cta() elif not shown: print(fallback) if not self.agent: _credits_or(t("cli.shared.no_active_agent")) return agent = self.agent calls = agent.session_api_calls if calls == 0: _credits_or(t("cli.usage.no_api_calls")) return rl_state = agent.get_rate_limit_state() if rl_state and rl_state.has_data: from agent.rate_limit_tracker import format_rate_limit_display print() print(format_rate_limit_display(rl_state)) print() input_tokens = getattr(agent, "session_input_tokens", 0) or 0 output_tokens = getattr(agent, "session_output_tokens", 0) or 0 reasoning_tokens = getattr(agent, "session_reasoning_tokens", 0) or 0 compressor = agent.context_compressor last_prompt = compressor.last_prompt_tokens if compressor.last_prompt_tokens > 0 else 0 ctx_len = compressor.context_length pct = min(100, (last_prompt / ctx_len * 100)) if ctx_len else 0 elapsed = format_duration_compact((datetime.now() - self.session_start).total_seconds()) def _label_row(key: str, value: str) -> None: # Labels are re-padded here (not in the catalog) so translated widths still align. print(f" {t(key):<26} {value}") print(f" {t('cli.usage.header_session')}") print(f" {'─' * 40}") _label_row("cli.usage.label_model", str(agent.model)) _label_row("cli.usage.label_input_tokens", f"{input_tokens:>10,}") _label_row("cli.usage.label_output_tokens", f"{output_tokens:>10,}") if reasoning_tokens: _label_row("cli.usage.label_reasoning_subset", f"{reasoning_tokens:>10,}") _label_row("cli.usage.label_prompt_tokens_total", f"{agent.session_prompt_tokens:>10,}") _label_row("cli.usage.label_completion_tokens", f"{agent.session_completion_tokens:>10,}") _label_row("cli.usage.label_total_tokens", f"{agent.session_total_tokens:>10,}") _label_row("cli.usage.label_api_calls", f"{calls:>10,}") _label_row("cli.usage.label_session_duration", f"{elapsed:>10}") print(f" {'─' * 40}") from agent.context_breakdown import context_display_source mark = "~" if context_display_source(compressor) != "provider_usage" else "" from agent.context_pin import context_pin_suffix pin_suffix = context_pin_suffix(ctx_len, getattr(agent, '_config_context_length', None)) print(f" {t('cli.usage.label_current_context'):<17} {mark}{last_prompt:,} / {ctx_len:,} ({mark}{pct:.0f}%){pin_suffix}") print(f" {t('cli.usage.label_messages'):<17} {len(self.conversation_history)}") print(f" {t('cli.usage.label_compressions'):<17} {compressor.compression_count}") self._print_account_limits() if self._print_nous_credits_block(): self._print_usage_cta() if self.verbose: logging.getLogger().setLevel(logging.DEBUG) for noisy in ('openai', 'openai._base_client', 'httpx', 'httpcore', 'asyncio', 'hpack', 'grpc', 'modal'): logging.getLogger(noisy).setLevel(logging.WARNING) else: logging.getLogger().setLevel(logging.INFO) def _print_account_limits(self) -> bool: """Provider account limits block for `/usage`; True if anything printed. Uses the live agent's route when present, else the CLI's own configured provider (the TUI/Desktop slash-worker runs without an agent). Fetched off-thread with a hard timeout so slow provider APIs don't hang the prompt; failures are non-fatal. Lazy import: pulls the OpenAI SDK chain. """ provider = self._agent_or_self("provider") if not provider: return False from agent.account_usage import fetch_account_usage, render_account_usage_lines account_snapshot = None with concurrent.futures.ThreadPoolExecutor(max_workers=1) as _pool: try: account_snapshot = _pool.submit( fetch_account_usage, provider, base_url=self._agent_or_self("base_url"), api_key=self._agent_or_self("api_key"), ).result(timeout=10.0) except (concurrent.futures.TimeoutError, Exception): account_snapshot = None account_lines = [f" {line}" for line in render_account_usage_lines(account_snapshot)] if not account_lines: return False print() for line in account_lines: print(line) return True def _show_insights(self, command: str = "/insights"): """Show usage insights and analytics from session history (`--days N` / `N`, `--source`).""" parts = command.split() days = 30 source = None i = 1 while i < len(parts): if parts[i] == "--days" and i + 1 < len(parts): try: days = int(parts[i + 1]) except ValueError: print(f" {t('gateway.insights.invalid_days', value=parts[i + 1])}") return i += 2 elif parts[i] == "--source" and i + 1 < len(parts): source = parts[i + 1] i += 2 else: if parts[i].isdigit(): days = int(parts[i]) i += 1 try: from hermes_state import SessionDB, _default_db_path from agent.insights import InsightsEngine if not _default_db_path().exists(): print(f" {t('cli.insights.no_session_data')}") return db = SessionDB(read_only=True) try: engine = InsightsEngine(db) print(engine.format_terminal(engine.generate(days=days, source=source))) finally: db.close() except Exception as e: print(f" {t('gateway.insights.error', error=str(e))}") def _check_config_mcp_changes(self) -> None: """Detect mcp_servers changes in config.yaml (polled from process_loop every CONFIG_WATCH_INTERVAL seconds) and react. Default (``mcp.auto_reload_on_config_change: true``) auto-triggers ``_reload_mcp()``. When opted out it only notifies and points at ``/reload-mcp`` — every reload rebuilds the tool surface and INVALIDATES the provider prompt cache (next message re-sends the full prefix), so silent reloads are wrong when external tooling rewrites config.yaml often. Instead it notifies the user that the config changed and that they can apply it with ``/reload-mcp`` — while warning that ``/reload-mcp`` rebuilds the tool surface and **invalidates the provider prompt cache** (the next message re-sends the full input prefix, expensive on long-context / high-reasoning models). See #1474. """ import hermes_yaml as _yaml now = time.monotonic() if now - self._last_config_check < CONFIG_WATCH_INTERVAL: return self._last_config_check = now from hermes_cli.config import get_config_path as _get_config_path cfg_path = _get_config_path() if not cfg_path.exists(): return try: sig = file_signature(cfg_path.stat()) except OSError: return if sig == self._config_sig: return # unchanged — fast path self._config_sig = sig try: with open(cfg_path, encoding="utf-8-sig") as f: new_cfg = _yaml.safe_load(f) or {} except Exception: return # Expand ${VAR} templates so the comparison matches the init snapshot (populated from the # deep-merged + expanded config); otherwise any save_config_value() rewrite of an # unrelated key would false-positive on "${POWERMEM_API_KEY}" vs its expanded value. from hermes_cli.config import _expand_env_vars new_mcp = _expand_env_vars(new_cfg.get("mcp_servers") or {}) if new_mcp == self._config_mcp_servers: return # some other section was edited # Read the toggle from the config just parsed so the user can flip it in the same edit; # missing key means default-on. _mcp_cfg = new_cfg.get("mcp") _auto = _mcp_cfg.get("auto_reload_on_config_change", True) if isinstance(_mcp_cfg, dict) else True self._config_mcp_servers = new_mcp if not _auto: print() print(t("cli.mcp_watch.changed_reload_skipped")) print(f" {t('cli.mcp_watch.not_applied_run')}") print(" /reload-mcp") print(f" {t('cli.mcp_watch.cache_note')}") return # Separate thread so a hung MCP server can't block process_loop (freezing the TUI). print() print(t("cli.mcp_watch.changed_reloading")) threading.Thread(target=self._reload_mcp, daemon=True).start() def _confirm_and_reload_mcp(self, cmd_original: str = "") -> None: """Interactive /reload-mcp — confirm (Approve Once / Always Approve / Cancel, gated by ``approvals.mcp_reload_confirm``, default on), then reload. The config watcher's auto-reload calls ``_reload_mcp`` directly. Reloading invalidates the provider prompt cache (tool schemas are baked into the system prompt), hence the warning.""" choice = _gated_confirm( self, "reload-mcp", "mcp_reload_confirm", title=t("cli.reload_mcp.title"), detail=t("cli.reload_mcp.detail"), choices=_reload_mcp_choices(), unchanged=t("cli.reload_mcp.unchanged"), always_msg=t("cli.reload_mcp.always_msg"), once_verb=t("cli.reload_mcp.once_verb")) if choice is None: return with self._busy_command(self._slow_command_status(cmd_original)): self._reload_mcp() def _reload_mcp(self): """Reload MCP servers: disconnect all, re-read config.yaml, reconnect, then refresh the agent's tool list so the model sees the updated tools on the next turn.""" try: from tools.mcp_tool_lifecycle import shutdown_mcp_servers from tools.mcp_tool_discovery import discover_mcp_tools from tools.mcp_tool_agent import reprobe_tool_availability from tools.mcp_tool import _servers, _lock with _lock: old_servers = set(_servers.keys()) if not self._command_running: print(t("cli.reload_mcp.reloading")) shutdown_mcp_servers() reprobe_tool_availability() # explicit reload also re-probes check_fn availability new_tools = discover_mcp_tools() # reads config.yaml fresh with _lock: connected_servers = set(_servers.keys()) # English labels are model-facing (the [IMPORTANT: …] note below); the printed lines # resolve their own localized key per bucket. diff = { "Added": connected_servers - old_servers, "Removed": old_servers - connected_servers, "Reconnected": connected_servers & old_servers} for label, key in (("Reconnected", "cli.reload_mcp.reconnected"), ("Added", "cli.reload_mcp.added"), ("Removed", "cli.reload_mcp.removed")): if diff[label]: print(f" {t(key, names=', '.join(sorted(diff[label])))}") if not connected_servers: print(f" {t('gateway.reload_mcp.none_connected')}") else: print(f" {t('cli.reload_mcp.tools_available', tools=str(len(new_tools)), servers=str(len(connected_servers)))}") # Route through the shared helper so this path stays in lockstep with the TUI RPC / # gateway reload / late-binding paths (name-diff, thread-safe, additive-preserving so # memory-provider and context-engine tools survive the rebuild). if self.agent is not None: from tools.mcp_tool_agent import refresh_agent_mcp_tools # Pick up servers ENABLED in config this session: enabled_toolsets was resolved at # startup, so merge now-connected names in (unless `all`/`*` is pinned) so a # freshly-added server isn't filtered out. Mirrors startup (see __init__). enabled_override = None et = self.enabled_toolsets if et and "all" not in et and "*" not in et: merged = list(et) for _name in sorted(connected_servers): if _name not in merged: merged.append(_name) enabled_override = merged refresh_agent_mcp_tools(self.agent, enabled_override=enabled_override, quiet_mode=True) if enabled_override is not None: self.enabled_toolsets = enabled_override # Tell the model tools changed — appended at the END so the prefix cache survives. change_parts = [ f"{label} servers: {', '.join(sorted(names))}" for label, names in diff.items() if names ] tool_summary = f"{len(new_tools)} MCP tool(s) now available" if new_tools else "No MCP tools available" change_detail = ". ".join(change_parts) + ". " if change_parts else "" self.conversation_history.append({ "role": "user", "content": f"[IMPORTANT: MCP servers have been reloaded. {change_detail}{tool_summary}. The tool list for this conversation has been updated accordingly.]", }) # Persist now so the session log reflects the refreshed tools list (best-effort). if self.agent is not None: try: self.agent._persist_session(self.conversation_history, self.conversation_history) except Exception: pass print(f" {t('cli.reload_mcp.agent_updated', count=str(len(self.agent.tools if self.agent else [])))}") except Exception as e: print(f" {t('gateway.reload_mcp.failed', error=str(e))}") def _reload_skills(self) -> None: """Reload skills: rescan ~/.hermes/skills/ and queue a note for the next user turn. Skills are invoked at runtime (``/skill-name``, ``skills_list``, ``skill_view``), not from the system prompt, so this does NOT clear the prompt cache. If anything was added/removed a one-shot note is prepended to the NEXT user message (``_pending_skills_reload_note``, same pattern as ``_pending_model_switch_note``) — nothing is written to conversation_history, so message alternation stays intact. """ try: from agent.skill_commands import reload_skills, get_skill_commands if not self._command_running: print(t("cli.reload_skills.reloading")) result = reload_skills() # Sync cli.py's module-level _skill_commands so help / dispatch / Tab-completion see # the updated dict without a restart. import cli as _cli _cli._skill_commands = get_skill_commands() added = result.get("added", []) # [{"name", "description"}, ...] removed = result.get("removed", []) total = result.get("total", 0) if not added and not removed: print(f" {t('gateway.reload_skills.no_new')}") print(f" {t('cli.reload_skills.total', count=str(total))}") return if added: print(f" {t('cli.reload_skills.added_header')}") for item in added: print(f" {_skill_line(item)}") if removed: print(f" {t('cli.reload_skills.removed_header')}") for item in removed: print(f" {_skill_line(item)}") print(f" {t('cli.reload_skills.total', count=str(total))}") # Same shape as the system prompt's skill catalog (`` - name: description``). sections = ["[USER INITIATED SKILLS RELOAD:"] if added: sections += ["", "Added Skills:", *(_skill_line(item) for item in added)] if removed: sections += ["", "Removed Skills:", *(_skill_line(item) for item in removed)] sections += ["", "Use skills_list to see the updated catalog.]"] self._pending_skills_reload_note = "\n".join(sections) except Exception as e: print(f" {t('gateway.reload_skills.failed', error=str(e))}")