diff --git a/hermes_cli/cli_info_mixin.py b/hermes_cli/cli_info_mixin.py index 87c3286c5e..28659f7bf9 100644 --- a/hermes_cli/cli_info_mixin.py +++ b/hermes_cli/cli_info_mixin.py @@ -1,8 +1,8 @@ -"""Informational views and reload flows for the interactive CLI: banner, help, tools, usage, insights, MCP/skills reload, bang shell +"""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 (``from cli import ...``) — the mixin -never imports ``cli`` at module load time (import cycle). +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 @@ -18,9 +18,51 @@ from hermes_constants import is_termux as _is_termux_environment from rich.markup import escape as _escape from utils import base_url_hostname +from hermes_cli.cli_modal_mixin import _gated_confirm + +CONFIG_WATCH_INTERVAL = 5.0 # seconds between config.yaml stat() calls + +_TOOL_PROGRESS_CYCLE = ["off", "new", "all", "verbose"] + +_RELOAD_MCP_CHOICES = [ + ("once", "Approve Once", "reload now"), + ("always", "Always Approve", "reload now and silence this prompt permanently"), + ("cancel", "Cancel", "leave MCP tools unchanged"), +] +_RELOAD_MCP_DETAIL = ( + "Reloading MCP servers rebuilds the tool set for this session and\n" + "invalidates the provider prompt cache. The next message will\n" + "re-send full input tokens (can be expensive on long-context or\n" + "high-reasoning models)." +) + + +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 = {t["function"]["name"]: get_toolset_for_tool(t["function"]["name"]) for t 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""" + """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.""" @@ -29,52 +71,39 @@ class CLIInfoMixin: ctx_len = None if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'): ctx_len = self.agent.context_compressor.context_length - - # Auto-compact for narrow terminals — the full banner with caduceus - # + tool list needs ~80 columns minimum to render without wrapping. - term_width = shutil.get_terminal_size().columns - use_compact = self.compact or term_width < 80 - - if use_compact: + + # 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 below - # re-verifies the snapshot so any drift self-heals next launch. + # 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, + compute_toolset_availability, load_banner_snapshot, save_banner_snapshot, ) - - snapshot = None try: snapshot = load_banner_snapshot(self.enabled_toolsets) except Exception: snapshot = None - - # Get terminal working directory (where commands will execute) - cwd = os.getenv("TERMINAL_CWD", os.getcwd()) + 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, + ) if snapshot is not None: self._defer_tool_warnings = True toolset_map = snapshot["toolset_map"] build_welcome_banner( - console=self.console, - model=self.model, - cwd=cwd, tools=snapshot["tools"], - enabled_toolsets=self.enabled_toolsets, - session_id=self.session_id, get_toolset_for_tool=lambda name: toolset_map.get(name), - context_length=ctx_len, - provider=self.provider, availability=snapshot["availability"], skills_by_category=snapshot.get("skills_by_category"), + **banner_kw, ) def _refresh_banner_snapshot() -> None: @@ -84,61 +113,29 @@ class CLIInfoMixin: enabled_toolsets=self.enabled_toolsets, quiet_mode=True ) availability = compute_toolset_availability(self.enabled_toolsets) - tmap = { - t["function"]["name"]: get_toolset_for_tool(t["function"]["name"]) - for t in tools - } - for item in availability.get("unavailable_toolsets", []): - for name in item.get("tools", []): - tmap.setdefault( - name, item.get("id", item.get("name", "")) - ) - save_banner_snapshot( - tools, self.enabled_toolsets, availability, tmap - ) + 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, + target=_refresh_banner_snapshot, name="banner-snapshot-refresh", daemon=True, ).start() else: - # Cold path: compute everything live, then persist the snapshot - # so the next launch replays it. + # 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, quiet_mode=True) availability = compute_toolset_availability(self.enabled_toolsets) - - build_welcome_banner( - console=self.console, - model=self.model, - cwd=cwd, - tools=tools, - enabled_toolsets=self.enabled_toolsets, - session_id=self.session_id, - context_length=ctx_len, - provider=self.provider, - availability=availability, - ) + build_welcome_banner(tools=tools, availability=availability, **banner_kw) try: - tmap = { - t["function"]["name"]: get_toolset_for_tool(t["function"]["name"]) - for t in tools - } - for item in availability.get("unavailable_toolsets", []): - for name in item.get("tools", []): - tmap.setdefault(name, item.get("id", item.get("name", ""))) + 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 intentionally deferred on the Termux bare prompt - # path; availability warnings are shown once tools are initialized. - # On the snapshot fast path (warm launch), the check walks every - # check_fn (~180ms) — run it in the background refresh thread instead - # and let its output land above the prompt (patch_stdout-safe). + + # 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( @@ -149,8 +146,7 @@ class CLIInfoMixin: else: self._show_tool_availability_warnings() - # Warn about low context lengths (common with local servers). Keep - # this tied to the runtime guard so guidance cannot drift again. + # Low context warning — tied to the runtime guard so guidance cannot drift. from agent.model_metadata import MINIMUM_CONTEXT_LENGTH if ctx_len and ctx_len < MINIMUM_CONTEXT_LENGTH: self._console_print() @@ -164,12 +160,10 @@ class CLIInfoMixin: base_url = getattr(self, "base_url", "") or "" from urllib.parse import urlparse as _urlparse try: - _parsed = _urlparse(base_url if "://" in base_url else f"//{base_url}") - _port = _parsed.port + _port = _urlparse(base_url if "://" in base_url else f"//{base_url}").port except ValueError: _port = None - _host = base_url_hostname(base_url) - if _port == 11434 or "ollama" in _host: + if _port == 11434 or "ollama" in base_url_hostname(base_url): self._console_print( f"[dim] Ollama fix: OLLAMA_CONTEXT_LENGTH={MINIMUM_CONTEXT_LENGTH} ollama serve[/]" ) @@ -182,11 +176,8 @@ class CLIInfoMixin: "[dim] Fix: Set model.context_length in config.yaml, or increase your server's context setting[/]" ) - # Warn if the configured model is a Nous Hermes LLM (not agentic) from hermes_cli.model_switch import is_nous_hermes_non_agentic - - model_name = getattr(self, "model", "") or "" - if is_nous_hermes_non_agentic(model_name): + if is_nous_hermes_non_agentic(getattr(self, "model", "") or ""): self._console_print() self._console_print( "[bold yellow]⚠ Nous Research Hermes 3 & 4 models are NOT agentic and are not " @@ -196,28 +187,19 @@ class CLIInfoMixin: "[dim] They lack tool-calling capabilities required for agent workflows. " "Consider using an agentic model (Claude, GPT, Gemini, DeepSeek, etc.).[/]" ) - self._console_print( - "[dim] Switch with: /model sonnet or /model gpt5[/]" - ) + self._console_print("[dim] Switch with: /model sonnet or /model gpt5[/]") - # Project-local skills: one-line status. Trusted → show count; - # untrusted-with-skills → point at `hermes skills trust`. Never raises. + # 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, + 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 - ) + _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]◆ {_n} project skill(s) loaded from this repo[/]" - ) + self._console_print(f"[dim]◆ {_n} project skill(s) loaded from this repo[/]") else: _untrusted = get_untrusted_project_skills_root() if _untrusted is not None: @@ -237,8 +219,7 @@ class CLIInfoMixin: except Exception: return False agent = getattr(self, "agent", None) - model = getattr(agent, "model", None) or getattr(self, "model", None) - return model_supports_fast_mode(model) + 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": @@ -246,37 +227,31 @@ class CLIInfoMixin: 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 commands by substring. - """ + """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, + ChatConsole, _BOLD, _DIM, _RST, _accent_hex, _cprint, _ensure_skill_commands, + _termux_example_image_path, get_skill_bundles, ) from hermes_cli.commands import COMMANDS_BY_CATEGORY, HELP_SESSION_SUBGROUPS arg = (arg or "").strip() skill_commands = _ensure_skill_commands() - # /help skills — the full skill-command list (kept out of the default - # view so core commands don't scroll off screen). + 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"): if not skill_commands: _cprint("\n No skill commands installed.\n") return _cprint(f"\n ⚡ {_BOLD}Skill Commands{_RST} ({len(skill_commands)} installed):") for cmd, info in sorted(skill_commands.items()): - ChatConsole().print( - f" [bold {_accent_hex()}]{cmd:<22}[/] [dim]-[/] {_escape(info['description'])}" - ) + _row(cmd, info['description'], 22) _cprint("") return @@ -287,66 +262,42 @@ class CLIInfoMixin: header = get_active_help_header("(^_^)? Available Commands") except Exception: header = "(^_^)? Available Commands" - header = (header or "").strip() or "(^_^)? Available Commands" - inner_width = 55 - if len(header) > inner_width: - header = header[:inner_width] - _cprint(f"\n{_BOLD}+{'-' * inner_width}+{_RST}") - _cprint(f"{_BOLD}|{header:^{inner_width}}|{_RST}") - _cprint(f"{_BOLD}+{'-' * inner_width}+{_RST}") - - def _emit(cmd: str, desc: str) -> bool: - if not self._command_available(cmd): - return False - if query and query not in cmd.lower() and query not in desc.lower(): - return False - ChatConsole().print( - f" [bold {_accent_hex()}]{cmd:<15}[/] [dim]-[/] {_escape(desc)}" - ) - return True - - for category, commands in COMMANDS_BY_CATEGORY.items(): - if category == "Session": - # Split the oversized Session category into readable sub-groups - # (Session / Context / Background & Automation) in the renderer. - sub_of: dict[str, str] = {} - for _sub, _names in HELP_SESSION_SUBGROUPS.items(): - for _n in _names: - sub_of[f"/{_n}"] = _sub - buckets: dict[str, list[tuple[str, str]]] = {"Session": []} - for _sub in HELP_SESSION_SUBGROUPS: - buckets[_sub] = [] - for cmd, desc in commands.items(): - buckets[sub_of.get(cmd, "Session")].append((cmd, desc)) - for _sub in ("Session", *HELP_SESSION_SUBGROUPS.keys()): - rows = buckets.get(_sub) or [] - 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}── {_sub} ──{_RST}") - printed_header = True - _emit(cmd, desc) - continue + header = ((header or "").strip() or "(^_^)? Available Commands")[: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 commands.items(): + 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}── {category} ──{_RST}") + _cprint(f"\n {_BOLD}── {title} ──{_RST}") printed_header = True - _emit(cmd, desc) + _row(cmd, desc) - # Skill commands: collapsed to a one-line pointer by default so the - # 60+ skill entries don't bury the core command reference (C-04). + for category, commands in COMMANDS_BY_CATEGORY.items(): + if category != "Session": + _section(category, commands.items()) + 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.items(): + 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: - # In filter mode, DO include matching skill commands inline. matched_skills = [ (cmd, info) for cmd, info in sorted(skill_commands.items()) if query in cmd.lower() or query in (info.get("description", "").lower()) @@ -354,9 +305,7 @@ class CLIInfoMixin: if matched_skills: _cprint(f"\n ⚡ {_BOLD}Skill Commands{_RST} (matching '{arg}'):") for cmd, info in matched_skills: - ChatConsole().print( - f" [bold {_accent_hex()}]{cmd:<22}[/] [dim]-[/] {_escape(info['description'])}" - ) + _row(cmd, info['description'], 22) elif skill_commands: _cprint( f"\n ⚡ {_BOLD}Skill Commands{_RST}: {len(skill_commands)} installed " @@ -378,10 +327,7 @@ class CLIInfoMixin: if quick_commands and not query: _cprint(f"\n ⚡ {_BOLD}Quick Commands{_RST} ({len(quick_commands)} configured):") for name, qcmd in sorted(quick_commands.items()): - desc = qcmd.get("description", qcmd.get("type", "")) - ChatConsole().print( - f" [bold {_accent_hex()}]{('/' + name):<22}[/] [dim]-[/] {_escape(desc)}" - ) + _row('/' + name, qcmd.get("description", qcmd.get("type", "")), 22) if query: _cprint(f"\n {_DIM}Filtered by '{arg}' — run /help for the full list.{_RST}\n") @@ -398,47 +344,34 @@ class CLIInfoMixin: def show_tools(self): """Display available tools with kawaii ASCII art.""" from cli import get_tool_definitions, get_toolset_for_tool - # Pre-assembly list: /tools is a discovery/inspection surface, so it - # must show the full catalog including tools deferred behind the - # tool_search bridge (users check this to verify an MCP installed). + # 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, quiet_mode=True, skip_tool_search_assembly=True) - if not tools: print("(;_;) No tools available") return - - # Header + print() - title = "(^_^)/ Available Tools" - width = 78 - pad = width - len(title) - print("+" + "-" * width + "+") - print("|" + " " * (pad // 2) + title + " " * (pad - pad // 2) + "|") - print("+" + "-" * width + "+") + _ascii_box("(^_^)/ Available Tools", 78) print() - - # Group tools by toolset - toolsets = {} + + toolsets: dict[str, list] = {} for tool in sorted(tools, key=lambda t: t["function"]["name"]): name = tool["function"]["name"] toolset = get_toolset_for_tool(name) or "unknown" - if toolset not in toolsets: - toolsets[toolset] = [] - desc = tool["function"].get("description", "") - # First sentence: split on ". " (period+space) to avoid breaking on "e.g." or "v2.0" - desc = desc.split("\n")[0] + 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[toolset].append((name, desc)) - - # Display by toolset + 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" Total: {len(tools)} tools ヽ(^o^)ノ") print() @@ -446,27 +379,17 @@ class CLIInfoMixin: """Display available toolsets with kawaii ASCII art.""" from cli import get_all_toolsets, get_toolset_info all_toolsets = get_all_toolsets() - - # Header + print() - title = "(^_^)b Available Toolsets" - width = 58 - pad = width - len(title) - print("+" + "-" * width + "+") - print("|" + " " * (pad // 2) + title + " " * (pad - pad // 2) + "|") - print("+" + "-" * width + "+") + _ascii_box("(^_^)b Available Toolsets", 58) print() - + for name in sorted(all_toolsets.keys()): info = get_toolset_info(name) if info: - tool_count = info["tool_count"] - desc = info["description"] - - # Mark if currently enabled marker = "(*)" if self.enabled_toolsets and name in self.enabled_toolsets else " " - print(f" {marker} {name:<18} [{tool_count:>2} tools] - {desc}") - + print(f" {marker} {name:<18} [{info['tool_count']:>2} tools] - {info['description']}") + print() print(" (*) = currently enabled") print() @@ -477,12 +400,10 @@ class CLIInfoMixin: 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(" You: cli (local terminal)") print(f" User: {user_name}") @@ -490,17 +411,14 @@ class CLIInfoMixin: print(" Slash commands: all available") print() - def _should_handle_steer_command_inline(self, text: str, has_images: bool = False) -> bool: - """Return True when /steer should be dispatched immediately while the agent is running. + 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. - /steer MUST bypass the normal _pending_input → process_loop path when - the agent is active, because process_loop is blocked inside - self.chat() for the duration of the run. By the time the queued - command is pulled from _pending_input, _agent_running has already - flipped back to False, and process_command() takes the idle - fallback — delivering the steer as a next-turn message instead of - injecting it mid-run. Dispatching inline on the UI thread calls - agent.steer() directly, which is thread-safe (uses _pending_steer_lock). + 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): @@ -509,82 +427,46 @@ class CLIInfoMixin: return False try: from hermes_cli.commands import resolve_command - base = text.split(None, 1)[0].lower().lstrip('/') - cmd = resolve_command(base) - return bool(cmd and cmd.name == "steer") + 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 should be dispatched immediately while the agent is running.""" + return self._busy_inline_command(text, has_images, ("steer",)) + 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. - - Same queue problem /steer had. ``/bg`` exists to start independent - work *without* waiting for the current turn, and ``/btw`` exists to - answer a side question about the in-flight conversation, but a slash - command typed while the agent is busy goes into ``_pending_input``, - and ``process_loop`` is blocked inside ``self.chat()`` for the whole - run. The side task would therefore only start once the foreground - turn has finished, which is the one moment it was not needed. - - Both commands' ``CommandDef`` entries already declare - ``busy_policy="dispatch"``; the gateway honours that, the classic CLI - never consulted it. Dispatching inline on the UI thread starts the - side session immediately and leaves the foreground turn running - untouched: no interrupt, no steer. - """ - 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 - base = text.split(None, 1)[0].lower().lstrip('/') - cmd = resolve_command(base) - return bool(cmd and cmd.name in ("bg", "btw")) - except Exception: - return False + """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-command routing and before - anything is queued for the agent, so a bang command never becomes a - turn: no user message, no assistant message, no tool result touches - ``self.conversation_history``. That is what makes ``!`` free — zero - tokens, and role alternation / prompt caching are untouched by - construction. The invariant is covered by - tests/cli/test_bang_shell_mode.py. - - Returns False when the text is not a bang command or when bang mode is - disabled for this context (gateway/cron), letting the caller fall - through to normal routing. + 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/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, + 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 contexts: no composer, no human at a - # keyboard, and those users already have their own shells. Let the - # text route normally rather than becoming remote execution. + # 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 instead of running an - # empty shell or sending "!" to the model. + if not command: # bare `!` — show what the feature does self._console_print(f"[dim]{USAGE_HINT}[/]") return True @@ -596,10 +478,9 @@ class CLIInfoMixin: self._console_print(f"[bold red]{_escape(str(message))}[/]") return True - cwd = resolve_bang_cwd(getattr(self, "session_id", None)) exit_code = run_bang_command( command, - cwd=cwd, + cwd=resolve_bang_cwd(getattr(self, "session_id", None)), writer=lambda line: self._console_print(_rich_text_from_ansi(line)), ) if exit_code: @@ -610,26 +491,23 @@ class CLIInfoMixin: """Show status of the gateway and connected messaging platforms.""" from cli import display_hermes_home from gateway.config import load_gateway_config, Platform - + print() print("+" + "-" * 60 + "+") print("|" + " " * 15 + "(✿◠‿◠) Gateway Status" + " " * 17 + "|") print("+" + "-" * 60 + "+") print() - + try: config = load_gateway_config() - print(" Messaging Platform Configuration:") 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: @@ -638,7 +516,7 @@ class CLIInfoMixin: print(f" ✓ {name:<12} Enabled{home_str}") else: print(f" ○ {name:<12} Not configured ({env_var})") - + print() print(" Session Reset Policy:") print(" " + "-" * 55) @@ -646,14 +524,12 @@ class CLIInfoMixin: print(f" Mode: {policy.mode}") print(f" Daily reset at: {policy.at_hour}:00") print(f" Idle timeout: {policy.idle_minutes} minutes") - print() print(" To start the gateway:") print(" python cli.py --gateway") print() print(f" Configuration file: {display_hermes_home()}/config.yaml") print() - except Exception as e: print(f" Error loading gateway config: {e}") print() @@ -681,25 +557,20 @@ class CLIInfoMixin: def _toggle_verbose(self): """Cycle tool progress mode: off → new → all → verbose → off. - Tool-progress display (full args / results / think blocks at the - ``verbose`` step) is INDEPENDENT of global DEBUG logging. Cycling - through here does not change ``self.verbose`` or the agent's - ``verbose_logging`` / ``quiet_mode`` — those remain under the - explicit ``-v``/``--verbose`` flag and the ``/verbose-logging`` - toggle. See PR #6a1aa420e for the history that decoupled them. + 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 - cycle = ["off", "new", "all", "verbose"] try: - idx = cycle.index(self.tool_progress_mode) + idx = _TOOL_PROGRESS_CYCLE.index(self.tool_progress_mode) except ValueError: idx = 2 # default to "all" - self.tool_progress_mode = cycle[(idx + 1) % len(cycle)] + 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. Leaving _focus_view_enabled - # set would show a "focus" status-bar badge and hidden-line counts while - # tool lines were visibly printing. Display-only state change. + # /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 @@ -707,22 +578,17 @@ class CLIInfoMixin: 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() - # Keep the live agent's tool_progress_mode in sync so the - # tool_executor rendering path reflects the new mode this turn, - # without waiting for an agent rebuild. + # Sync the live agent so tool_executor rendering reflects the new mode this turn. self.agent.tool_progress_mode = self.tool_progress_mode - # Use raw ANSI codes via _cprint so the output is routed through - # prompt_toolkit's renderer. self.console.print() with Rich markup - # writes directly to stdout which patch_stdout's StdoutProxy mangles - # into garbled sequences like '?[33mTool progress: NEW?[0m' (#2262). + # Raw ANSI via _cprint so output routes through prompt_toolkit's renderer; Rich markup to + # stdout gets mangled by patch_stdout's StdoutProxy ('?[33mTool progress: NEW?[0m'). from hermes_cli.colors import Colors as _Colors labels = { "off": f"{_Colors.DIM}Tool progress: OFF{_Colors.RESET} — silent mode, just the final response.", @@ -733,12 +599,8 @@ class CLIInfoMixin: _cprint(labels.get(self.tool_progress_mode, "")) def _handle_usage_command(self, cmd_original: str): - """Dispatch `/usage [reset [--force]]`. - - Bare `/usage` keeps the classic display. `/usage reset` redeems one - banked Codex rate-limit reset credit (guarded: refuses when limits - aren't exhausted unless --force). - """ + """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": @@ -749,20 +611,16 @@ class CLIInfoMixin: 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.""" - provider = ( - (getattr(self.agent, "provider", None) if self.agent else None) - or getattr(self, "provider", None) - ) - normalized = str(provider or "").strip().lower() - if normalized != "openai-codex": + if str(self._agent_or_self("provider") or "").strip().lower() != "openai-codex": print(" Banked usage resets are only available on the openai-codex provider.") print(" Switch with `/model` or `hermes auth` first.") return - base_url = (getattr(self.agent, "base_url", None) if self.agent else None) or getattr(self, "base_url", None) - api_key = (getattr(self.agent, "api_key", None) if self.agent else None) or getattr(self, "api_key", None) - from agent.account_usage import redeem_codex_reset_credit print(" ⏳ Checking banked reset credits...") @@ -770,8 +628,8 @@ class CLIInfoMixin: try: result = _pool.submit( redeem_codex_reset_credit, - base_url=base_url, - api_key=api_key, + 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: @@ -780,18 +638,9 @@ class CLIInfoMixin: print(f" {result.message}") def _show_context_breakdown(self, cmd_original: str = ""): - """`/context [all]` — visual context-window usage breakdown. - - Renders a 5×20 glyph block grid (each cell ≈ 1% of the model context - window) plus an estimated per-category table: system prompt, tool - definitions, rules, skills index, MCP, subagents, memory, and the - conversation itself — versus free space. `/context all` appends the - expanded per-skill and per-toolset cost listings. - - Read-only: same chars/4 estimation engine as the desktop context - popover (agent.context_breakdown) — no provider calls, no prompt-cache - impact. - """ + """`/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(" (._.) No active agent -- send a message first.") return @@ -800,15 +649,11 @@ class CLIInfoMixin: expanded = args in {"all", "full", "details"} from agent.context_breakdown import ( - compute_context_details, - compute_session_context_breakdown, + compute_context_details, compute_session_context_breakdown, render_context_breakdown_lines, ) - try: - payload = compute_session_context_breakdown( - self.agent, self.conversation_history - ) + payload = compute_session_context_breakdown(self.agent, self.conversation_history) except Exception as e: print(f" (._.) Could not compute context breakdown: {e}") return @@ -820,9 +665,8 @@ class CLIInfoMixin: except Exception: details = {"skills": [], "toolsets": []} - model = payload.get("model") or self.model print() - print(f" 🧠 Context Usage — {model}") + print(f" 🧠 Context Usage — {payload.get('model') or self.model}") print() for line in render_context_breakdown_lines(payload, details=details, grid=True): print(f" {line}") @@ -831,30 +675,26 @@ class CLIInfoMixin: def _show_usage(self): """Rate limits + session token usage (when a live agent exists) + Nous credits. - The Nous credits block is agent-independent (a portal fetch), so it runs even - with no live agent — important for the TUI, where /usage runs in a slash-worker - subprocess that resumes the session WITHOUT building an agent (self.agent is None), - which would otherwise early-return before any credits showed. + 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 - if not self.agent: + + def _credits_or(fallback: str) -> None: if self._print_nous_credits_block(): self._print_usage_cta() else: - print("(._.) No active agent -- send a message first.") - return + print(fallback) + if not self.agent: + _credits_or("(._.) No active agent -- send a message first.") + return agent = self.agent calls = agent.session_api_calls - if calls == 0: - if self._print_nous_credits_block(): - self._print_usage_cta() - else: - print("(._.) No API calls made yet in this session.") + _credits_or("(._.) No API calls made yet in this session.") return - # ── Rate limits (shown first when available) ──────────────── 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 @@ -862,21 +702,13 @@ class CLIInfoMixin: print(format_rate_limit_display(rl_state)) print() - # ── Session token usage ───────────────────────────────────── 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 - prompt = agent.session_prompt_tokens - completion = agent.session_completion_tokens - total = agent.session_total_tokens - 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 - compressions = compressor.compression_count - - msg_count = len(self.conversation_history) elapsed = format_duration_compact((datetime.now() - self.session_start).total_seconds()) print(" 📊 Session Token Usage") @@ -886,22 +718,19 @@ class CLIInfoMixin: print(f" Output tokens: {output_tokens:>10,}") if reasoning_tokens: print(f" ↳ Reasoning (subset): {reasoning_tokens:>10,}") - print(f" Prompt tokens (total): {prompt:>10,}") - print(f" Completion tokens: {completion:>10,}") - print(f" Total tokens: {total:>10,}") + print(f" Prompt tokens (total): {agent.session_prompt_tokens:>10,}") + print(f" Completion tokens: {agent.session_completion_tokens:>10,}") + print(f" Total tokens: {agent.session_total_tokens:>10,}") print(f" API calls: {calls:>10,}") print(f" Session duration: {elapsed:>10}") print(f" {'─' * 40}") print(f" Current context: {last_prompt:,} / {ctx_len:,} ({pct:.0f}%)") - print(f" Messages: {msg_count}") - print(f" Compressions: {compressions}") + print(f" Messages: {len(self.conversation_history)}") + print(f" Compressions: {compressor.compression_count}") - # Account limits -- fetched off-thread with a hard timeout so slow - # provider APIs don't hang the prompt. - provider = getattr(agent, "provider", None) or getattr(self, "provider", None) - base_url = getattr(agent, "base_url", None) or getattr(self, "base_url", None) - api_key = getattr(agent, "api_key", None) or getattr(self, "api_key", None) - # Lazy import — pulls the OpenAI SDK chain, only needed here. + # Account limits — fetched off-thread with a hard timeout so slow provider APIs don't + # hang the prompt. Lazy import: pulls the OpenAI SDK chain. + provider = self._agent_or_self("provider") from agent.account_usage import fetch_account_usage, render_account_usage_lines account_snapshot = None if provider: @@ -909,7 +738,8 @@ class CLIInfoMixin: try: account_snapshot = _pool.submit( fetch_account_usage, provider, - base_url=base_url, api_key=api_key, + 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 @@ -919,8 +749,6 @@ class CLIInfoMixin: for line in account_lines: print(line) - # Nous credits magnitudes + monthly-grant gauge (agent-independent — also - # runs at the no-agent / no-calls early-returns above). See the helper. if self._print_nous_credits_block(): self._print_usage_cta() @@ -932,8 +760,7 @@ class CLIInfoMixin: logging.getLogger().setLevel(logging.INFO) def _show_insights(self, command: str = "/insights"): - """Show usage insights and analytics from session history.""" - # Parse optional --days flag + """Show usage insights and analytics from session history (`--days N` / `N`, `--source`).""" parts = command.split() days = 30 source = None @@ -949,51 +776,34 @@ class CLIInfoMixin: elif parts[i] == "--source" and i + 1 < len(parts): source = parts[i + 1] i += 2 - elif parts[i].isdigit(): - days = int(parts[i]) - i += 1 else: + if parts[i].isdigit(): + days = int(parts[i]) i += 1 try: from hermes_state import SessionDB from agent.insights import InsightsEngine - db = SessionDB() try: engine = InsightsEngine(db) - report = engine.generate(days=days, source=source) - print(engine.format_terminal(report)) + print(engine.format_terminal(engine.generate(days=days, source=source))) finally: db.close() except Exception as e: print(f" Error generating insights: {e}") def _check_config_mcp_changes(self) -> None: - """Detect mcp_servers changes in config.yaml and react. + """Detect mcp_servers changes in config.yaml (polled from process_loop every + CONFIG_WATCH_INTERVAL seconds) and react. - Called from process_loop every CONFIG_WATCH_INTERVAL seconds. - Compares config.yaml mtime + mcp_servers section against the last - known state. When a change is detected: - - * By default (``mcp.auto_reload_on_config_change: true``) it - auto-triggers ``_reload_mcp()`` and informs the user — legacy - behaviour from #1474. - * When opted out (``mcp.auto_reload_on_config_change: false``) it - does NOT reload. 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). This stops silent cache-breaking reloads - when config.yaml is rewritten frequently by external tooling or - other Hermes instances. + 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. """ - import yaml as _yaml - CONFIG_WATCH_INTERVAL = 5.0 # seconds between config.yaml stat() calls - now = time.monotonic() if now - self._last_config_check < CONFIG_WATCH_INTERVAL: return @@ -1003,16 +813,13 @@ class CLIInfoMixin: cfg_path = _get_config_path() if not cfg_path.exists(): return - try: mtime = cfg_path.stat().st_mtime except OSError: return - if mtime == self._config_mtime: - return # File unchanged — fast path + return # unchanged — fast path - # File changed — check whether mcp_servers section changed self._config_mtime = mtime try: with open(cfg_path, encoding="utf-8") as f: @@ -1020,43 +827,21 @@ class CLIInfoMixin: except Exception: return - new_mcp = new_cfg.get("mcp_servers") or {} - # Expand ${VAR} templates so the comparison is consistent with the - # init snapshot (self._config_mcp_servers), which was populated from - # the deep-merged + expanded config. Without this, any - # save_config_value() that rewrites config.yaml (even for unrelated - # keys) triggers a false-positive MCP reload because the raw yaml - # still has "${POWERMEM_API_KEY}" while the snapshot has the - # expanded value. + # 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_mcp) + new_mcp = _expand_env_vars(new_cfg.get("mcp_servers") or {}) if new_mcp == self._config_mcp_servers: - return # mcp_servers unchanged (some other section was edited) + return # some other section was edited - # Detected a change in the mcp_servers section. By default we - # auto-reload (legacy behaviour), but if the user has opted out we - # notify instead of reloading — because every reload rebuilds the - # agent tool surface and INVALIDATES the provider prompt cache (the - # next message re-sends the full input prefix, which is expensive on - # long-context / high-reasoning models). - # - # The toggle is the top-level ``mcp.auto_reload_on_config_change`` - # key (see DEFAULT_CONFIG). Read it from the config we just parsed - # so the user can flip it in the same edit that changes mcp_servers; + # 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 - ) - + _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: - # Notify the user that the config changed but do NOT auto-reload. - # They can apply the new settings on their own terms with - # /reload-mcp — which we explicitly warn may invalidate the cache. print() print("🔄 MCP server config changed — reload skipped (auto-reload disabled).") print(" New settings are NOT applied yet. To apply them now, run:") @@ -1065,117 +850,48 @@ class CLIInfoMixin: print(" provider prompt cache (next message re-sends full input tokens).") return - # Notify user and reload. Run in a separate thread with a hard - # timeout so a hung MCP server cannot block the process_loop - # indefinitely (which would freeze the entire TUI). + # Separate thread so a hung MCP server can't block process_loop (freezing the TUI). print() print("🔄 MCP server config changed — reloading connections...") - _reload_thread = threading.Thread( - target=self._reload_mcp, daemon=True - ) - _reload_thread.start() + threading.Thread(target=self._reload_mcp, daemon=True).start() def _confirm_and_reload_mcp(self, cmd_original: str = "") -> None: - """Interactive /reload-mcp — confirm with the user, then reload. - - The auto-reload path (config file watcher) calls ``_reload_mcp`` - directly and never goes through this confirmation. - - Reloading MCP tools invalidates the provider prompt cache for the - active session (tool schemas are baked into the system prompt). - The next message re-sends full input tokens — can be expensive on - long-context or high-reasoning models. - - Three options: Approve Once, Always Approve (persists - ``approvals.mcp_reload_confirm: false`` so future reloads run - without this prompt), Cancel. Gated by - ``approvals.mcp_reload_confirm`` — default on. - """ - from cli import load_cli_config, save_config_value - # Gate check — respects prior "Always Approve" clicks. - try: - cfg = load_cli_config() - approvals = cfg.get("approvals") if isinstance(cfg, dict) else None - confirm_required = True - if isinstance(approvals, dict): - confirm_required = bool(approvals.get("mcp_reload_confirm", True)) - except Exception: - confirm_required = True - - if not confirm_required: - with self._busy_command(self._slow_command_status(cmd_original)): - self._reload_mcp() - return - - # Render warning + prompt. Use the same prompt_toolkit-native composer - # modal as destructive slash confirmations so choices stay visible. - choices = [ - ("once", "Approve Once", "reload now"), - ("always", "Always Approve", "reload now and silence this prompt permanently"), - ("cancel", "Cancel", "leave MCP tools unchanged"), - ] - raw = self._prompt_text_input_modal( + """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="⚠️ /reload-mcp — Prompt cache invalidation warning", - detail=( - "Reloading MCP servers rebuilds the tool set for this session and\n" - "invalidates the provider prompt cache. The next message will\n" - "re-send full input tokens (can be expensive on long-context or\n" - "high-reasoning models)." - ), - choices=choices, + detail=_RELOAD_MCP_DETAIL, + choices=_RELOAD_MCP_CHOICES, + unchanged="MCP tools unchanged.", + always_msg="🔒 Future /reload-mcp calls will run without confirmation.", + once_verb="reloading", ) - if raw is None: - print("🟡 /reload-mcp cancelled (no input).") - return - choice = self._normalize_slash_confirm_choice(raw, choices) if choice is None: - print(f"🟡 Unrecognized choice '{raw}'. /reload-mcp cancelled.") return - - if choice == "cancel": - print("🟡 /reload-mcp cancelled. MCP tools unchanged.") - return - - if choice == "always": - if save_config_value("approvals.mcp_reload_confirm", False): - print("🔒 Future /reload-mcp calls will run without confirmation.") - print(" Re-enable via `approvals.mcp_reload_confirm: true` in config.yaml.") - else: - print("⚠️ Couldn't persist opt-out — reloading once.") - 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. - - After reconnecting, refreshes the agent's tool list so the model - sees the updated tools on the next turn. - """ + """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 import ( shutdown_mcp_servers, discover_mcp_tools, reprobe_tool_availability, _servers, _lock, ) - - # Capture old server names with _lock: old_servers = set(_servers.keys()) - if not self._command_running: print("🔄 Reloading MCP servers...") - # Shutdown existing connections shutdown_mcp_servers() + reprobe_tool_availability() # explicit reload also re-probes check_fn availability + new_tools = discover_mcp_tools() # reads config.yaml fresh - # Explicit reload also re-probes tool availability (check_fn). - reprobe_tool_availability() - # Reconnect (reads config.yaml fresh) - new_tools = discover_mcp_tools() - - # Compute what changed with _lock: connected_servers = set(_servers.keys()) - added = connected_servers - old_servers removed = old_servers - connected_servers reconnected = connected_servers & old_servers @@ -1191,19 +907,14 @@ class CLIInfoMixin: else: print(f" 🔧 {len(new_tools)} tool(s) available from {len(connected_servers)} server(s)") - # Refresh the agent's tool list so the model can call new tools. - # Route through the shared helper so this CLI /reload-mcp path stays - # in lockstep with the TUI RPC / gateway reload / late-binding paths - # (name-diff, thread-safe, and — critically — additive-preserving so + # 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 import refresh_agent_mcp_tools - # Explicit reload: pick up MCP servers the user ENABLED in config - # this session. self.enabled_toolsets was resolved once at - # startup; merge in any now-connected server names (unless the - # user pinned `all`/`*`, which already includes everything) so a - # freshly-added server isn't filtered out. Mirrors startup, where - # MCP server names are part of enabled_toolsets (see __init__). + # 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: @@ -1212,18 +923,11 @@ class CLIInfoMixin: if _name not in merged: merged.append(_name) enabled_override = merged - refresh_agent_mcp_tools( - self.agent, - enabled_override=enabled_override, - quiet_mode=True, - ) - # Keep the CLI's own list in sync with what the agent now uses. + refresh_agent_mcp_tools(self.agent, enabled_override=enabled_override, quiet_mode=True) if enabled_override is not None: self.enabled_toolsets = enabled_override - # Inject a message at the END of conversation history so the - # model knows tools changed. Appended after all existing - # messages to preserve prompt-cache for the prefix. + # Tell the model tools changed — appended at the END so the prefix cache survives. change_parts = [] if added: change_parts.append(f"Added servers: {', '.join(sorted(added))}") @@ -1238,50 +942,38 @@ class CLIInfoMixin: "content": f"[IMPORTANT: MCP servers have been reloaded. {change_detail}{tool_summary}. The tool list for this conversation has been updated accordingly.]", }) - # Persist session immediately so the session log reflects the - # updated tools list (self.agent.tools was refreshed above). + # 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, - ) + self.agent._persist_session(self.conversation_history, self.conversation_history) except Exception: - pass # Best-effort + pass print(f" ✅ Agent updated — {len(self.agent.tools if self.agent else [])} tool(s) available") - except Exception as e: print(f" ❌ MCP reload failed: {e}") def _reload_skills(self) -> None: - """Reload skills: rescan ~/.hermes/skills/ and queue a note for the - next user turn. + """Reload skills: rescan ~/.hermes/skills/ and queue a note for the next user turn. - Skills don't need to live in the system prompt for the model to use - them (they're invoked via ``/skill-name``, ``skills_list``, or - ``skill_view`` at runtime), so this does NOT clear the prompt cache. - It rescans the slash-command map, prints the diff for the user, and - — if any skills were added or removed — queues a one-shot note that - gets prepended to the next user message. This preserves message - alternation (no phantom user turn injected out of band) and keeps - prompt caching intact. + 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("🔄 Reloading skills...") - result = reload_skills() - # Sync cli.py's module-level _skill_commands so all consumers - # (help display, command dispatch, Tab-completion lambda) see the - # updated dict without needing to restart the session. + # 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", []) # [{"name", "description"}, ...] + removed = result.get("removed", []) total = result.get("total", 0) if not added and not removed: @@ -1289,45 +981,23 @@ class CLIInfoMixin: print(f" 📚 {total} skill(s) available") return - def _fmt_line(item: dict) -> str: - nm = item.get("name", "") - desc = item.get("description", "") - return f" - {nm}: {desc}" if desc else f" - {nm}" - if added: print(" ➕ Added Skills:") for item in added: - print(f" {_fmt_line(item)}") + print(f" {_skill_line(item)}") if removed: print(" ➖ Removed Skills:") for item in removed: - print(f" {_fmt_line(item)}") + print(f" {_skill_line(item)}") print(f" 📚 {total} skill(s) available") - # Queue a one-shot note for the NEXT user turn. The CLI's agent - # loop prepends ``_pending_skills_reload_note`` (if set) to the - # API-call-local message at ~L8770, then clears it — same - # pattern as ``_pending_model_switch_note``. Nothing is written - # to conversation_history here, so message alternation stays - # intact and no out-of-band user turn is persisted. - # - # Format matches how the system prompt renders pre-existing - # skills (`` - name: description``) so the model reads the - # diff in the same shape as its original skill catalog. + # Same shape as the system prompt's skill catalog (`` - name: description``). sections = ["[USER INITIATED SKILLS RELOAD:"] if added: - sections.append("") - sections.append("Added Skills:") - for item in added: - sections.append(_fmt_line(item)) + sections += ["", "Added Skills:", *(_skill_line(item) for item in added)] if removed: - sections.append("") - sections.append("Removed Skills:") - for item in removed: - sections.append(_fmt_line(item)) - sections.append("") - sections.append("Use skills_list to see the updated catalog.]") + 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" ❌ Skills reload failed: {e}") diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index 9b8c6abb90..80130366e5 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -1,8 +1,8 @@ -"""Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command palette, slash-confirm, external editor +"""Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command palette, +slash-confirm, external editor. -Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. cli.py-internal -symbols are imported LAZILY inside each method (``from cli import ...``) — the mixin -never imports ``cli`` at module load time (import cycle). +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 @@ -10,13 +10,79 @@ from __future__ import annotations import json import queue import sys +import threading +import time as _time from hermes_cli.callbacks import prompt_for_secret from typing import Optional +_TIMED_OUT = object() # sentinel returned by _poll_modal_queue when the deadline passes + +# Typed answers accepted by the slash-confirm modal, mapped onto the canonical choice values. +_CONFIRM_ALIASES = { + "1": "once", "once": "once", "approve": "once", "yes": "once", "y": "once", "ok": "once", + "2": "always", "always": "always", "remember": "always", + "3": "cancel", "cancel": "cancel", "nevermind": "cancel", "no": "cancel", "n": "cancel", +} + +_APPROVAL_OUTCOME_LABELS = { + "once": "allowed once", + "session": "allowed for session", + "always": "added to allowlist", + "deny": "denied", +} + +_CLARIFY_TIMEOUT_REPLY = ( + "The user did not provide a response within the time limit. " + "Use your best judgement to make the choice and proceed." +) + + +def _approval_gate_on(key: str) -> bool: + """Read ``approvals.`` (default on); any load failure keeps the prompt enabled.""" + from cli import load_cli_config + try: + cfg = load_cli_config() + approvals = cfg.get("approvals") if isinstance(cfg, dict) else None + if isinstance(approvals, dict): + return bool(approvals.get(key, True)) + except Exception: + pass + return True + + +def _gated_confirm(self, command, key, *, title, detail, choices, unchanged, always_msg, once_verb): + """Shared once/always/cancel confirm behind ``approvals.`` (destructive slash, /reload-mcp). + + Returns ``"once"`` without prompting when the gate is off; ``None`` on cancel / no input / + unrecognized answer (already reported to the user). Picking "always" persists the opt-out. + """ + from cli import save_config_value + if not _approval_gate_on(key): + return "once" + raw = self._prompt_text_input_modal(title=title, detail=detail, choices=choices) + if raw is None: + print(f"🟡 /{command} cancelled (no input).") + return None + choice = self._normalize_slash_confirm_choice(raw, choices) + if choice is None: + print(f"🟡 Unrecognized choice '{raw}'. /{command} cancelled.") + return None + if choice == "cancel": + print(f"🟡 /{command} cancelled. {unchanged}") + return None + if choice == "always": + if save_config_value(f"approvals.{key}", False): + print(always_msg) + print(f" Re-enable via `approvals.{key}: true` in config.yaml.") + else: + print(f"⚠️ Couldn't persist opt-out — {once_verb} once.") + return choice + class CLIModalMixin: - """Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command palette, slash-confirm, external editor""" + """Modal overlays for the interactive CLI: clarify, approval, sudo/secret capture, command + palette, slash-confirm, external editor.""" def _open_external_editor(self, buffer=None) -> bool: """Open the active input buffer in an external editor.""" @@ -28,7 +94,8 @@ class CLIModalMixin: if self._command_running: _cprint(f"{_DIM}Wait for the current command to finish before opening the editor.{_RST}") return False - if self._sudo_state or self._secret_state or self._approval_state or getattr(self, "_slash_confirm_state", None) or self._clarify_state: + if (self._sudo_state or self._secret_state or self._approval_state + or getattr(self, "_slash_confirm_state", None) or self._clarify_state): _cprint(f"{_DIM}Finish the active prompt before opening the editor.{_RST}") return False target_buffer = buffer or getattr(app, "current_buffer", None) @@ -36,38 +103,27 @@ class CLIModalMixin: _cprint(f"{_DIM}No active input buffer is available for the external editor.{_RST}") return False try: - # Inline pastes so the editor (and the draft it submits) sees real - # content; skip flag unconditionally so the editor-close text-change - # doesn't re-collapse it, even when there was nothing to inline. + # Inline pastes so the editor sees real content; set the skip flag unconditionally so + # the editor-close text-change doesn't re-collapse it. self._inline_pastes(target_buffer) self._skip_paste_collapse = True - # Open the editor, then submit the saved draft on a clean exit — - # matching the TUI's Ctrl+G (openEditor), which sends the buffer - # instead of requiring a second Enter. Submission in this CLI is - # driven by the custom `enter` keybinding, NOT the buffer's - # accept_handler, so validate_and_handle can't route through it; - # chain a done-callback on the returned Task that re-uses the - # real submit pipeline via _submit_editor_buffer(). + # Submission here is driven by the custom `enter` keybinding, NOT the buffer's + # accept_handler, so validate_and_handle can't route through it; chain a done-callback + # that re-uses the real submit pipeline (TUI Ctrl+G parity: save == send). task = target_buffer.open_in_editor(validate_and_handle=False) if task is not None and hasattr(task, "add_done_callback"): - task.add_done_callback( - lambda _t, b=target_buffer: self._submit_editor_buffer(b) - ) + task.add_done_callback(lambda _t, b=target_buffer: self._submit_editor_buffer(b)) return True except Exception as exc: _cprint(f"{_DIM}Failed to open external editor: {exc}{_RST}") return False def _submit_editor_buffer(self, buffer) -> None: - """Submit the draft an external editor left in ``buffer``. + """Submit the draft an external editor left in ``buffer`` (Ctrl+G done-callback). - Invoked from the Ctrl+G done-callback so saving the editor sends the - prompt (TUI parity) instead of leaving it sitting in the input area. - Mirrors the idle/queue branches of the `enter` keybinding handler: - an empty save is ignored (never submits a blank turn), a slash command - is dispatched, otherwise the text is routed through the same input - queues the normal Enter path uses. Runs on the prompt_toolkit event - loop via the Task callback, so it must be cheap and non-blocking. + Mirrors the idle/queue branches of the `enter` keybinding: an empty save is ignored (never + submits a blank turn), a bang/slash command is dispatched, otherwise the text goes through + the same input queues. Runs on the prompt_toolkit loop, so it must stay cheap/non-blocking. """ from cli import _DIM, _RST, _cprint, _looks_like_slash_command try: @@ -75,30 +131,25 @@ class CLIModalMixin: except Exception: return if not text: - # Editor saved empty / was cleared — match the TUI, which drops - # an empty draft instead of submitting a blank turn. return app = getattr(self, "_app", None) - # `!` shell mode, checked before slash dispatch — matches the - # Enter path in the input loop so an editor-saved bang command runs - # locally instead of being sent to the agent. - try: - if self.handle_bang_shell(text): - self._reset_input_buffer(buffer) - if app is not None: - app.invalidate() - return - except Exception as exc: - _cprint(f" {_DIM}Shell command failed: {exc}{_RST}") + def _done() -> None: self._reset_input_buffer(buffer) if app is not None: app.invalidate() + + # `!` shell mode is checked before slash dispatch, matching the Enter path. + try: + if self.handle_bang_shell(text): + _done() + return + except Exception as exc: + _cprint(f" {_DIM}Shell command failed: {exc}{_RST}") + _done() return - # Slash commands: dispatch directly, same as the Enter handler's - # _looks_like_slash_command branch. if _looks_like_slash_command(text): try: if not self.process_command(text): @@ -108,35 +159,28 @@ class CLIModalMixin: except Exception as exc: _cprint(f" {_DIM}Command failed: {exc}{_RST}") finally: - self._reset_input_buffer(buffer) - if app is not None: - app.invalidate() + _done() return - # Regular prompt: route through the same queues the Enter handler uses. if self._agent_running: - # Agent busy → honour the configured busy-input behaviour by - # queueing for the next turn (the safe default; interrupt/steer - # remain reachable via the normal Enter path). - self._interrupt_queue.put(text) if self.busy_input_mode == "interrupt" else self._pending_input.put(text) + # Agent busy → honour the configured busy-input behaviour (interrupt/steer remain + # reachable via the normal Enter path). + if self.busy_input_mode == "interrupt": + self._interrupt_queue.put(text) + else: + self._pending_input.put(text) preview = text[:80] + ("..." if len(text) > 80 else "") _cprint(f" Queued for the next turn: {preview}") else: self._pending_input.put(text) - - self._reset_input_buffer(buffer) - if app is not None: - app.invalidate() + _done() def _inline_pastes(self, buffer) -> None: - """Replace collapsed-paste placeholders in ``buffer`` with real content. + """Replace collapsed ``[Pasted text #N -> file]`` placeholders in ``buffer`` with real text. - A big paste shows as a compact ``[Pasted text #N -> file]`` placeholder, - but history recall and the external editor need the actual text — a bare - reference is useless once the file is gone or on another machine. Inlining - before ``reset(append_to_history=True)`` also lets prompt_toolkit persist - the content through its normal path. Sets ``_skip_paste_collapse`` so the - ensuing text-change doesn't re-collapse it. + History recall and the external editor need the content (the file may be gone or on another + machine); inlining before ``reset(append_to_history=True)`` also lets prompt_toolkit persist + it. Sets ``_skip_paste_collapse`` so the ensuing text-change doesn't re-collapse it. """ from cli import logger try: @@ -178,14 +222,11 @@ class CLIModalMixin: def _prompt_text_input(self, prompt_text: str) -> str | None: """Prompt for free-text input safely inside or outside prompt_toolkit. - ``run_in_terminal`` returns a coroutine that must be awaited by the prompt_toolkit event loop, - which only exists on the main thread. Slash commands are dispatched from - the ``process_loop`` daemon thread (see issue #23185), so calling - ``run_in_terminal`` from there orphans the coroutine — ``_ask`` never runs, - and user keystrokes leak into the composer instead. Fall back to a direct - ``input()`` when we're off the main thread. + ``run_in_terminal`` returns a coroutine only the main-thread event loop can await; slash + commands run on the ``process_loop`` daemon thread, where a bare ``input()`` would block + forever on loop-owned stdin (TUI slash-worker hang). Off the main thread with an app running + we therefore cancel cleanly (None) — mirroring ``_stdin_fallback`` in the modal prompt. """ - import threading result = [None] def _ask(): @@ -195,14 +236,6 @@ class CLIModalMixin: pass in_main_thread = threading.current_thread() is threading.main_thread() - - # Slash-worker guard (#23185 / billing auto-reload hang): when a - # prompt_toolkit app is running but we're on a non-main thread (the - # process_loop / TUI slash-worker daemon thread), stdin is owned by the - # event loop / JSON-RPC pipe. A bare input() there blocks forever until - # the worker's 45s timeout fires. We cannot safely prompt off the main - # thread, so cancel cleanly (None) instead of hanging — mirrors the - # _stdin_fallback discipline in _prompt_text_input_modal. if self._app and not in_main_thread: self._invalidate() return None @@ -215,9 +248,8 @@ class CLIModalMixin: try: run_in_terminal(_ask) except Exception: - # WSL / Warp / certain terminal emulators silently drop the - # scheduled coroutine. Fall back to a direct input() so the - # user's keystrokes don't leak into the agent buffer. + # WSL / Warp / some emulators silently drop the scheduled coroutine — fall back to + # a direct input() so keystrokes don't leak into the agent buffer. try: _ask() except Exception: @@ -229,6 +261,28 @@ class CLIModalMixin: _ask() return result[0] + def _poll_modal_queue(self, response_queue, deadline_attr, *, refresh=1.0, paint=None): + """Block until a value lands on ``response_queue`` or ``self.`` passes + (``None`` deadline = unlimited). Returns the value or ``_TIMED_OUT``. + + Repaints every ``refresh`` seconds (``0`` = on every idle tick) so countdown hints stay + live; ``paint`` defaults to ``_paint_now`` — modal prompts must bypass the ``_invalidate`` + throttle/resize guard or the panel can be dropped and time out unseen. + """ + paint = paint or self._paint_now + last = _time.monotonic() + while True: + try: + return response_queue.get(timeout=1) + except queue.Empty: + deadline = getattr(self, deadline_attr) + if deadline is not None and deadline - _time.monotonic() <= 0: + return _TIMED_OUT + now = _time.monotonic() + if now - last >= refresh: + last = now + paint() + def _prompt_text_input_modal( self, *, @@ -237,38 +291,17 @@ class CLIModalMixin: choices: list[tuple[str, str, str]], timeout: float = 120, ) -> str | None: - """Prompt through the prompt_toolkit composer instead of raw input(). + """Slash-command confirmation through the prompt_toolkit composer instead of raw input(). - This is for CLI slash-command confirmations. The old raw input() path - fought prompt_toolkit's active stdin ownership: in some terminals the - prompt appeared above the TUI, choices were redrawn later, and Enter - could be interpreted as EOF/exit. A first-class modal state keeps the - choices visible and lets the normal Enter key binding submit the typed - or highlighted choice. - - **Platform note (Windows — issue #33961):** - Earlier code bypassed the modal on ``sys.platform == "win32"`` and fell - back to a raw ``input()`` prompt. When the confirm was triggered from the - ``process_loop`` daemon thread (the normal case) that ``input()`` ran off - the main thread and deadlocked against prompt_toolkit's stdin ownership — - the user saw a frozen cursor and Ctrl-C was swallowed (bare ``/reset`` - froze; ``/reset now`` worked only because it skips the prompt entirely). - - Native Windows now uses the same path as Linux/macOS: the modal is set up - on ``self._app.loop`` via ``call_soon_threadsafe`` and answered by the - normal prompt_toolkit key bindings (the same input channel that already - handles ordinary typing on Windows). The raw ``input()`` fallback is kept - only for the genuinely safe cases: no running app (unit tests / - non-interactive), no resolvable event loop, or a scheduling failure. + Raw input() fought prompt_toolkit's stdin ownership (prompt drawn above the TUI, Enter read + as EOF). The modal state keeps the choices visible and the normal Enter binding submits. + All platforms (incl. native Windows) drive the modal via ``self._app.loop`` + + ``call_soon_threadsafe``; the raw ``input()`` fallback is kept only for the safe cases: no + running app (tests / non-interactive), no resolvable loop, or a scheduling failure. On + Windows a non-main-thread input() deadlocks against prompt_toolkit, so that case cancels. """ - import threading - import time as _time - if not choices: return None - - # If prompt_toolkit is not running (unit tests / non-interactive calls), - # keep the simple stdin fallback. if not getattr(self, "_app", None): return self._prompt_text_input("Choice [1/2/3]: ") @@ -276,14 +309,9 @@ class CLIModalMixin: app_loop = self._app.loop except Exception: app_loop = None - in_main_thread = threading.current_thread() is threading.main_thread() def _stdin_fallback() -> str | None: - # On native Windows a raw input() from a non-main thread deadlocks - # against prompt_toolkit's stdin ownership (#33961). With an app - # running we cannot safely prompt off the main thread, so cancel - # cleanly (None) rather than hang the terminal. if sys.platform == "win32" and not in_main_thread: self._invalidate() return None @@ -332,22 +360,13 @@ class CLIModalMixin: if not _run_on_app_loop(_setup_modal): return _stdin_fallback() - - _last_countdown_refresh = _time.monotonic() try: - while True: - try: - result = response_queue.get(timeout=1) - _run_on_app_loop(_teardown_modal) - return result - except queue.Empty: - remaining = self._slash_confirm_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 5.0: - _last_countdown_refresh = now - self._invalidate() + result = self._poll_modal_queue( + response_queue, "_slash_confirm_deadline", refresh=5.0, paint=self._invalidate, + ) + if result is not _TIMED_OUT: + _run_on_app_loop(_teardown_modal) + return result finally: if self._slash_confirm_state is not None: _run_on_app_loop(_teardown_modal) @@ -372,24 +391,8 @@ class CLIModalMixin: choice_raw = raw.strip().lower() if not choice_raw: return None - aliases = { - "1": "once", - "once": "once", - "approve": "once", - "yes": "once", - "y": "once", - "ok": "once", - "2": "always", - "always": "always", - "remember": "always", - "3": "cancel", - "cancel": "cancel", - "nevermind": "cancel", - "no": "cancel", - "n": "cancel", - } allowed = {choice[0] for choice in choices} - normalized = aliases.get(choice_raw) + normalized = _CONFIRM_ALIASES.get(choice_raw) if normalized in allowed: return normalized if choice_raw in allowed: @@ -397,22 +400,17 @@ class CLIModalMixin: return None def _build_command_palette_entries(self) -> list: - """Flat list of (command, description) for the Ctrl+P palette. - - Sourced from the same COMMAND_REGISTRY that backs /help, filtered to - commands available on this surface, plus installed skill commands. - Selecting an entry inserts the exact command string — never a fuzzy - resolution. - """ + """Flat (command, category, desc) rows for the Ctrl+P palette: the COMMAND_REGISTRY behind + /help filtered to this surface, plus installed skill commands. Selecting inserts the exact + command string — never a fuzzy resolution.""" from cli import _ensure_skill_commands from hermes_cli.commands import COMMANDS_BY_CATEGORY - entries: list[tuple[str, str, str]] = [] # (command, category, desc) + entries: list[tuple[str, str, str]] = [] for category, commands in COMMANDS_BY_CATEGORY.items(): for cmd, desc in commands.items(): - if not self._command_available(cmd): - continue - entries.append((cmd, category, desc)) + if self._command_available(cmd): + entries.append((cmd, category, desc)) try: for cmd, info in sorted(_ensure_skill_commands().items()): entries.append((cmd, "Skill", info.get("description", ""))) @@ -421,10 +419,9 @@ class CLIModalMixin: return entries def _open_command_palette(self) -> None: - """Open the Ctrl+P fuzzy command palette modal.""" + """Open the Ctrl+P fuzzy command palette modal (never stacked over another modal).""" if getattr(self, "_command_palette_state", None): return - # Don't stack over other modals. if (self._model_picker_state or self._clarify_state or self._approval_state or self._slash_confirm_state or self._sudo_state or self._secret_state): return @@ -443,18 +440,10 @@ class CLIModalMixin: self._invalidate(min_interval=0.0) def _command_palette_visible_entries(self) -> list: - """Return (command, category, desc) rows matching the active filter. - - Ranked, command-name-focused matching (a bare subsequence over the - whole "cmd category desc" string is uselessly permissive — "steer" - would match 130+ rows via description text). Priority: - 0 exact command match - 1 command startswith query - 2 query substring in command - 3 query subsequence in command - 4 query substring in description - Rows that match nowhere are dropped. Ties keep registry order. - """ + """Rows matching the active filter, ranked command-name-first (a bare subsequence over + "cmd category desc" is uselessly permissive — "steer" would match 130+ rows via text): + 0 exact command, 1 command startswith, 2 substring in command, 3 subsequence in command, + 4 substring in description. Non-matches are dropped; ties keep registry order.""" state = self._command_palette_state or {} entries = state.get("entries") or [] q = (state.get("filter", "") or "").strip().lower() @@ -465,11 +454,11 @@ class CLIModalMixin: it = iter(hay) return all(ch in it for ch in needle) + qn = q.lstrip("/") ranked = [] for order, row in enumerate(entries): cmd, _cat, desc = row name = cmd.lower().lstrip("/") - qn = q.lstrip("/") desc_l = (desc or "").lower() if name == qn: rank = 0 @@ -488,7 +477,7 @@ class CLIModalMixin: return [row for (_r, _o, row) in ranked] def _handle_command_palette_selection(self) -> None: - """Insert the selected command into the composer (does not auto-run).""" + """Prefill the selected command into the composer — never auto-run (many take args).""" from cli import logger state = self._command_palette_state if not state: @@ -498,11 +487,8 @@ class CLIModalMixin: if not (0 <= selected < len(rows)): self._close_command_palette() return - cmd = rows[selected][0] # exact command string, e.g. "/model" + cmd = rows[selected][0] self._close_command_palette() - # Prefill the composer so the user can add args / confirm — never - # auto-execute (a palette pick should be explicit, and many commands - # take arguments). try: app = getattr(self, "_app", None) if app is not None: @@ -515,34 +501,19 @@ class CLIModalMixin: @classmethod def _split_destructive_skip(cls, cmd_text: Optional[str]) -> tuple[str, bool]: - """Split inline-skip tokens out of a destructive slash command. + """Split inline-skip tokens out of a destructive slash command → ``(remainder, skip)``. - Returns ``(remainder, skip)`` where ``remainder`` is the original - text with the command word and any recognized skip tokens removed, - and ``skip`` is True iff at least one skip token was found. - - Examples: - "/reset now" -> ("", True) - "/reset --yes My title" -> ("My title", True) - "/new My title" -> ("My title", False) - "/clear" -> ("", False) + ``remainder`` is the text minus the leading "/cmd" word and any skip tokens; ``skip`` is + True iff one was found: "/reset now" -> ("", True); "/reset --yes My title" -> + ("My title", True); "/new My title" -> ("My title", False). """ - if not cmd_text: - return "", False - tokens = cmd_text.strip().split() + tokens = (cmd_text or "").strip().split() if not tokens: return "", False - # Drop leading "/cmd" word — callers pass the full command text. if tokens[0].startswith("/"): tokens = tokens[1:] - skip = False - kept: list[str] = [] - for tok in tokens: - if tok.lower() in cls._DESTRUCTIVE_SKIP_TOKENS: - skip = True - continue - kept.append(tok) - return " ".join(kept), skip + kept = [tok for tok in tokens if tok.lower() not in cls._DESTRUCTIVE_SKIP_TOKENS] + return " ".join(kept), len(kept) != len(tokens) def _confirm_destructive_slash( self, @@ -550,100 +521,37 @@ class CLIModalMixin: detail: str, cmd_original: Optional[str] = None, ) -> Optional[str]: - """Prompt the user to confirm a destructive session slash command. + """Confirm a destructive session slash command (``/clear``, ``/new``/``/reset``, ``/undo``). - Used by ``/clear``, ``/new``/``/reset``, and ``/undo`` before they - discard conversation state. Three-option prompt: - - 1. Approve Once — proceed this time only - 2. Always Approve — proceed and persist - ``approvals.destructive_slash_confirm: false`` so future - destructive commands run without confirmation - 3. Cancel — abort - - Gated by ``approvals.destructive_slash_confirm`` (default on). If the - gate is off the function returns ``"once"`` immediately without - prompting. - - Inline-skip: if ``cmd_original`` contains ``now``, ``--yes``, or - ``-y`` as an argument (e.g. ``/reset now``, ``/new --yes My title``), - the modal is bypassed and ``"once"`` is returned immediately. This is - an escape hatch for non-interactive use and for the degraded path where - the modal can't be marshaled onto the app loop (native Windows itself now - drives the modal normally — see #33961). Callers are responsible - for stripping the skip tokens from any remaining argument parsing - (see :meth:`_split_destructive_skip`). - - Returns ``"once"``, ``"always"``, or ``None`` (cancelled). Callers - proceed with the destructive action when the result is non-None. + Returns ``"once"``, ``"always"`` (also persists ``approvals.destructive_slash_confirm: + false``) or ``None`` (cancelled); callers proceed when non-None. Gate off → ``"once"`` + without prompting. Inline-skip: ``now`` / ``--yes`` / ``-y`` in ``cmd_original`` bypasses + the modal (non-interactive escape hatch; callers strip the tokens via + :meth:`_split_destructive_skip`). """ - from cli import load_cli_config, save_config_value - # Inline-skip escape hatch — works regardless of platform/modal state. - # See class-level _DESTRUCTIVE_SKIP_TOKENS for the accepted tokens. - if cmd_original: - _, _skip = self._split_destructive_skip(cmd_original) - if _skip: - return "once" - - # Gate check — respects prior "Always Approve" clicks. - try: - cfg = load_cli_config() - approvals = cfg.get("approvals") if isinstance(cfg, dict) else None - confirm_required = True - if isinstance(approvals, dict): - confirm_required = bool(approvals.get("destructive_slash_confirm", True)) - except Exception: - confirm_required = True - - if not confirm_required: + if cmd_original and self._split_destructive_skip(cmd_original)[1]: return "once" - - # Render a prompt_toolkit-native confirmation panel. This keeps option - # labels visible above the composer and avoids raw input()/EOF races with - # the running TUI. - choices = [ - ("once", "Approve Once", "proceed this time only"), - ("always", "Always Approve", "proceed and silence this prompt permanently"), - ("cancel", "Cancel", "keep current conversation"), - ] - raw = self._prompt_text_input_modal( + return _gated_confirm( + self, command, "destructive_slash_confirm", title=f"⚠️ /{command} — destroys conversation state", detail=detail, - choices=choices, + choices=[ + ("once", "Approve Once", "proceed this time only"), + ("always", "Always Approve", "proceed and silence this prompt permanently"), + ("cancel", "Cancel", "keep current conversation"), + ], + unchanged="Conversation unchanged.", + always_msg="🔒 Future /clear, /new, /reset, and /undo will run without confirmation.", + once_verb="proceeding", ) - if raw is None: - print(f"🟡 /{command} cancelled (no input).") - return None - choice = self._normalize_slash_confirm_choice(raw, choices) - if choice is None: - print(f"🟡 Unrecognized choice '{raw}'. /{command} cancelled.") - return None - - if choice == "cancel": - print(f"🟡 /{command} cancelled. Conversation unchanged.") - return None - - if choice == "always": - if save_config_value("approvals.destructive_slash_confirm", False): - print("🔒 Future /clear, /new, /reset, and /undo will run without confirmation.") - print(" Re-enable via `approvals.destructive_slash_confirm: true` in config.yaml.") - else: - print("⚠️ Couldn't persist opt-out — proceeding once.") - - return choice def _ring_bell(self, prompt: bool = False, context: str = "", detail: str = "") -> None: """Write a terminal bell (\\a) if the matching display.bell_* flag is on. - ``prompt=True`` is the blocking-modal variant (clarify / approval / - sudo / secret capture) gated by ``display.bell_on_prompt``; the default - is the end-of-turn bell gated by ``display.bell_on_complete``. Works - over SSH — the BEL propagates to the user's terminal. - - The same flag also emits an OSC 9 desktop notification (Ghostty, - iTerm2, Kitty, WezTerm) and, inside a supporting Warp build, a - ``warp://cli-agent`` OSC 777 event — see ``hermes_cli.terminal_notify``. - ``context`` is the short notification body (e.g. "approval"). + ``prompt=True`` is the blocking-modal variant gated by ``display.bell_on_prompt``; the + default is the end-of-turn bell gated by ``display.bell_on_complete``. Works over SSH. The + same flag also emits the OSC 9 / Warp OSC 777 desktop notification via + ``hermes_cli.terminal_notify``; ``context`` is the short notification body. """ flag = "bell_on_prompt" if prompt else "bell_on_complete" if not getattr(self, flag, False): @@ -655,7 +563,6 @@ class CLIModalMixin: pass try: from hermes_cli.terminal_notify import notify as _terminal_notify - _terminal_notify( context or ("input needed" if prompt else "turn complete"), prompt=prompt, @@ -665,148 +572,96 @@ class CLIModalMixin: except Exception: pass - def _clarify_callback(self, question, choices, multi_select=False, questions=None): - """ - Platform callback for the clarify tool. Called from the agent thread. - - Sets up the interactive selection UI (or freetext prompt for open-ended - questions), then blocks until the user responds via the prompt_toolkit - key bindings. If no response arrives within the configured timeout the - question is dismissed and the agent is told to decide on its own. - - When ``multi_select`` is True, shows checkboxes and the user can - select multiple options with Space, confirming with Enter. - - When ``questions`` is a non-empty list (batch clarify, issue #18450), - the panel switches to the A-compact multi-question layout and the - return value is a dict ``{"answers": {qid: raw_answer}}`` (plus - ``"timed_out": True`` when the deadline expired with only partial - answers). The single-question path below is unchanged. - """ - from cli import CLI_CONFIG, _DIM, _RST, _cprint - import time as _time - - from tools.clarify_gateway import resolve_clarify_timeout - - if questions: - return self._clarify_callback_batch(questions) - - # Canonical clarify timeout, shared with the gateway/TUI path. `<= 0` - # means unlimited (never auto-skip mid-think) → a null deadline. - timeout = resolve_clarify_timeout(CLI_CONFIG) - response_queue = queue.Queue() - is_open_ended = not choices - # multi-select support: only active when multi_select is True and choices exist - effective_multi = multi_select and not is_open_ended - - self._clarify_state = { - "question": question, - "choices": choices if not is_open_ended else [], - "selected": 0, - # multi-select support - "multi_select": effective_multi, - "selected_indices": set() if effective_multi else None, - "response_queue": response_queue, - } - self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout - # Open-ended questions skip straight to freetext input - self._clarify_freetext = is_open_ended - self._clarify_multi_base = None - - self._ring_bell(prompt=True, context="clarify") - # Trigger an immediate prompt_toolkit repaint from this (non-main) - # thread. Modal prompts must paint at once and must not be gated by the - # _invalidate throttle / resize guard — see _paint_now / _invalidate (#41098). - self._paint_now() - - # Poll for the user's response. The countdown in the hint line updates - # on each repaint; refresh it once a second so the timer stays visible - # while we wait. Selection changes (↑/↓) trigger instant repaints via - # the key bindings. - _last_countdown_refresh = _time.monotonic() - while True: - try: - result = response_queue.get(timeout=1) - self._clarify_deadline = None - self._persist_prompt_summary("?", "Clarify", question, str(result)) - return result - except queue.Empty: - # None deadline = unlimited: never auto-skip, just keep polling. - if self._clarify_deadline is not None: - remaining = self._clarify_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 1.0: - _last_countdown_refresh = now - self._paint_now() - - # Timed out — tear down the UI and let the agent decide + def _clarify_teardown(self) -> None: self._clarify_state = None self._clarify_freetext = False self._clarify_deadline = None self._clarify_multi_base = None self._paint_now() + + def _clarify_callback(self, question, choices, multi_select=False, questions=None): + """Clarify-tool platform callback (agent thread): show the selection UI (or freetext for + open-ended questions) and block until the key bindings answer or the timeout dismisses it + (the agent is then told to decide). ``multi_select`` shows checkboxes (Space toggles). + A non-empty ``questions`` list switches to the batch panel and returns + ``{"answers": {qid: raw}}`` (plus ``"timed_out": True`` on a partial deadline expiry).""" + from cli import CLI_CONFIG, _DIM, _RST, _cprint + from tools.clarify_gateway import resolve_clarify_timeout + + if questions: + return self._clarify_callback_batch(questions) + + # Canonical clarify timeout, shared with the gateway/TUI path; `<= 0` = unlimited. + timeout = resolve_clarify_timeout(CLI_CONFIG) + response_queue = queue.Queue() + is_open_ended = not choices + effective_multi = multi_select and not is_open_ended + self._clarify_state = { + "question": question, + "choices": choices if not is_open_ended else [], + "selected": 0, + "multi_select": effective_multi, + "selected_indices": set() if effective_multi else None, + "response_queue": response_queue, + } + self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout + self._clarify_freetext = is_open_ended # open-ended → straight to freetext + self._clarify_multi_base = None + self._ring_bell(prompt=True, context="clarify") + self._paint_now() + + result = self._poll_modal_queue(response_queue, "_clarify_deadline") + if result is not _TIMED_OUT: + self._clarify_deadline = None + self._persist_prompt_summary("?", "Clarify", question, str(result)) + return result + self._clarify_teardown() _cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}") - return ( - "The user did not provide a response within the time limit. " - "Use your best judgement to make the choice and proceed." - ) + return _CLARIFY_TIMEOUT_REPLY def _clarify_batch_set_active(self, state, index) -> None: """Point the batch clarify panel at question ``index``. - Mirrors the active question's data into the flat keys the existing - single-question keybindings and renderer read (``question``, - ``choices``, ``selected``, ``multi_select``, ``selected_indices``), - so ↑/↓/Space/number keys operate on the active question unchanged. - Open-ended questions drop straight into freetext, matching the - single-question path. Re-visiting an answered question restores the - cursor to the earlier selection (choice answers highlight their row, - an "Other" answer highlights the Other row) so the user can see and - edit what they picked. + Mirrors the active question into the flat keys the single-question keybindings/renderer + read (``question``/``choices``/``selected``/``multi_select``/``selected_indices``) so + ↑/↓/Space/number keys work unchanged; open-ended drops into freetext. Re-visiting an + answered question restores the cursor/checkboxes to the earlier pick. """ questions_list = state["questions"] index = max(0, min(index, len(questions_list) - 1)) entry = questions_list[index] + choices = entry["choices"] or [] state["active"] = index state["question"] = entry["question"] - state["choices"] = entry["choices"] or [] + state["choices"] = choices state["selected"] = 0 state["multi_select"] = bool(entry["multi_select"]) state["selected_indices"] = set() if entry["multi_select"] else None self._clarify_freetext = not entry["choices"] self._clarify_multi_base = None - # Restore the earlier answer's cursor/checkbox position on re-visit. meta = (state.get("answer_meta") or {}).get(entry["qid"]) - choices = entry["choices"] or [] if meta is None: return - if meta.get("kind") == "choice": + kind = meta.get("kind") + if kind == "choice": answer = state["answers"].get(entry["qid"]) if answer in choices: state["selected"] = choices.index(answer) - elif meta.get("kind") == "other": + elif kind == "other": state["selected"] = len(choices) - elif meta.get("kind") == "multi": - checked = set() - for label in meta.get("choices") or []: - if label in choices: - checked.add(choices.index(label)) + elif kind == "multi": + checked = {choices.index(c) for c in meta.get("choices") or [] if c in choices} if meta.get("other_text"): checked.add(len(choices)) state["selected_indices"] = checked def _clarify_batch_lock(self, state, answer, meta=None) -> None: - """Lock ``answer`` for the active batch question and advance. + """Lock ``answer`` for the active batch question and advance to the next unanswered one. - Overwrites any earlier answer for the same question (locked answers - stay editable until the batch completes). ``meta`` records how the - answer was produced ({"kind": "choice"|"other"|"multi", ...}) so a - re-visit can restore the cursor and prefill an "Other" edit. Advances - ``active`` to the next unanswered question; when every question has - an answer, puts the answers dict on the response queue and tears down - the panel. + Overwrites an earlier answer (locked answers stay editable until the batch completes). + ``meta`` records how it was produced ({"kind": "choice"|"other"|"multi", ...}) so a + re-visit can restore the cursor / prefill an "Other" edit. When every question is answered + the answers dict goes on the response queue and the panel is torn down. """ entry = state["questions"][state["active"]] state["answers"][entry["qid"]] = answer @@ -818,7 +673,6 @@ class CLIModalMixin: if state["questions"][candidate]["qid"] not in state["answers"]: self._clarify_batch_set_active(state, candidate) return - # Every question answered — resolve the batch. try: state["response_queue"].put(dict(state["answers"])) except Exception: @@ -828,26 +682,21 @@ class CLIModalMixin: self._clarify_multi_base = None def _clarify_batch_enter(self, state) -> None: - """Enter in batch choice mode: lock the active question's selection. + """Enter in batch choice mode: lock the active selection. - Multi-select questions lock a JSON array string of the checked - labels (the tool core parses it via ``_parse_multi_select_response``). - Selecting "Other" switches to freetext; the freetext submit path - locks the typed answer. Entering "Other" on a question whose earlier - answer was typed prefills the composer with that text for editing. + Multi-select locks a JSON array of the checked labels (parsed by the tool core). "Other" + switches to freetext (the freetext submit locks the typed answer), prefilled with an + earlier typed answer so Enter on an answered Other edits instead of retyping. """ choices = state.get("choices") or [] selected = state.get("selected", 0) entry = state["questions"][state["active"]] meta = (state.get("answer_meta") or {}).get(entry["qid"]) or {} if state.get("multi_select"): - indices = state.get("selected_indices") or set() - sorted_idx = sorted(indices) + sorted_idx = sorted(state.get("selected_indices") or set()) selected_choices = [choices[i] for i in sorted_idx if i < len(choices)] - other_checked = len(choices) in sorted_idx - if other_checked: - # Stash the checked real choices (possibly none) so the - # freetext submit appends the typed answer to the array. + if len(choices) in sorted_idx: + # Stash the checked real choices so the freetext submit appends the typed answer. self._clarify_multi_base = selected_choices self._clarify_freetext = True self._clarify_prefill = meta.get("other_text") or "" @@ -859,42 +708,28 @@ class CLIModalMixin: ) return if selected < len(choices): - self._clarify_batch_lock( - state, choices[selected], meta={"kind": "choice"} - ) + self._clarify_batch_lock(state, choices[selected], meta={"kind": "choice"}) return - # "Other" highlighted → switch to freetext; prefill an earlier typed - # answer so Enter on an answered Other edits instead of retyping. self._clarify_freetext = True - self._clarify_prefill = ( - meta.get("other_text") or "" if meta.get("kind") == "other" else "" - ) + self._clarify_prefill = meta.get("other_text") or "" if meta.get("kind") == "other" else "" def _clarify_callback_batch(self, questions): - """Batch clarify panel (A-compact): all questions, one active. - - Blocks on the response queue like the single-question path. Returns - ``{"answers": {qid: raw_answer}}`` when every question is locked, the - same dict plus ``"timed_out": True`` when the deadline expires with - partial (or zero) answers, and passes a cancel string through - unchanged so the tool core resolves the batch empty. - """ + """Batch clarify panel (A-compact): all questions, one active. Returns + ``{"answers": {qid: raw}}`` when every question is locked, plus ``"timed_out": True`` when + the deadline expires with partial answers; a cancel string passes through unchanged so the + tool core resolves the batch empty.""" from cli import CLI_CONFIG, _DIM, _RST, _cprint - import time as _time - from tools.clarify_gateway import resolve_clarify_timeout timeout = resolve_clarify_timeout(CLI_CONFIG) response_queue = queue.Queue() - state = { "questions": list(questions), "answers": {}, "answer_meta": {}, "active": 0, "response_queue": response_queue, - # Flat keys mirroring the active question — filled by - # _clarify_batch_set_active below. + # Flat keys mirroring the active question — filled by _clarify_batch_set_active. "question": "", "choices": [], "selected": 0, @@ -907,112 +742,56 @@ class CLIModalMixin: self._ring_bell(prompt=True, context="clarify") self._paint_now() - _last_countdown_refresh = _time.monotonic() - while True: - try: - result = response_queue.get(timeout=1) - self._clarify_deadline = None - if isinstance(result, dict): - return {"answers": result} - # Cancel path (Ctrl+C teardown) posts a plain string — pass - # it through so the tool core resolves the batch empty. - return result - except queue.Empty: - if self._clarify_deadline is not None: - remaining = self._clarify_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 1.0: - _last_countdown_refresh = now - self._paint_now() - - # Timed out — keep the answers locked so far and flag the timeout. + result = self._poll_modal_queue(response_queue, "_clarify_deadline") + if result is not _TIMED_OUT: + self._clarify_deadline = None + return {"answers": result} if isinstance(result, dict) else result partial = dict(state["answers"]) - self._clarify_state = None - self._clarify_freetext = False - self._clarify_deadline = None - self._clarify_multi_base = None - self._paint_now() + self._clarify_teardown() _cprint(f"\n{_DIM}(clarify timed out after {timeout}s — locked answers returned){_RST}") return {"answers": partial, "timed_out": True} def _sudo_password_callback(self) -> str: - """ - Prompt for sudo password through the prompt_toolkit UI. - - Called from the agent thread when a sudo command is encountered. - Uses the same clarify-style mechanism: sets UI state, waits on a - queue for the user's response via the Enter key binding. - """ + """Prompt for a sudo password through the prompt_toolkit UI (agent thread); clarify-style + state + queue answered by the Enter binding.""" from cli import _DIM, _RST, _cprint - import time as _time - timeout = 45 response_queue = queue.Queue() - self._capture_modal_input_snapshot() - self._sudo_state = { - "response_queue": response_queue, - } - self._sudo_deadline = _time.monotonic() + timeout + self._sudo_state = {"response_queue": response_queue} + self._sudo_deadline = _time.monotonic() + 45 self._ring_bell(prompt=True, context="sudo password") - - # Modal prompt — paint immediately, bypassing the throttle/resize guard - # so the prompt can't be dropped and time out unseen (#41098). self._paint_now() - while True: - try: - result = response_queue.get(timeout=1) - self._sudo_state = None - self._sudo_deadline = 0 - self._restore_modal_input_snapshot() - self._paint_now() - if result: - _cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}") - else: - _cprint(f"\n{_DIM} ⏭ Skipped{_RST}") - return result - except queue.Empty: - remaining = self._sudo_deadline - _time.monotonic() - if remaining <= 0: - break - self._paint_now() - + result = self._poll_modal_queue(response_queue, "_sudo_deadline", refresh=0) self._sudo_state = None self._sudo_deadline = 0 self._restore_modal_input_snapshot() self._paint_now() - _cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}") - return "" + if result is _TIMED_OUT: + _cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}") + return "" + if result: + _cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}") + else: + _cprint(f"\n{_DIM} ⏭ Skipped{_RST}") + return result def _approval_callback(self, command: str, description: str, *, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False) -> str: - """ - Prompt for dangerous command approval through the prompt_toolkit UI. + """Dangerous-command approval through the prompt_toolkit UI (agent thread). - Called from the agent thread. Shows a selection UI similar to clarify - with choices: once / session / always / deny. Smart DENY owner - overrides show only once / deny, as do gates that re-ask every time - (allow_session=False). When allow_permanent is False for another - reason (for example tirith), only 'always' is hidden. - Long commands also get a 'view' option so the full command can be - expanded before deciding. - - Uses _approval_lock to serialize concurrent requests (e.g. from - parallel delegation subtasks) so each prompt gets its own turn - and the shared _approval_state / _approval_deadline aren't clobbered. + Choices: once / session / always / deny (see ``_approval_choices``), plus 'view' for long + commands. ``_approval_lock`` serializes concurrent requests (parallel delegation subtasks) + so the shared ``_approval_state`` / ``_approval_deadline`` aren't clobbered. """ from cli import CLI_CONFIG, _DIM, _RST, _cprint - import time as _time with self._approval_lock: timeout = int(CLI_CONFIG.get("approvals", {}).get("timeout", 300)) response_queue = queue.Queue() - self._approval_state = { "command": command, "description": description, @@ -1026,73 +805,40 @@ class CLIModalMixin: "response_queue": response_queue, } self._approval_deadline = _time.monotonic() + timeout - self._ring_bell(prompt=True, context="approval", detail=command) - # Modal prompt — paint immediately, bypassing the throttle/resize - # guard. A throttled paint here can be silently dropped (250ms - # window collision or in-flight resize), leaving the panel unseen so - # the command is denied on timeout without the user ever seeing it - # (#41098). The countdown refreshes below paint the same way. self._paint_now() - _last_countdown_refresh = _time.monotonic() - while True: - try: - result = response_queue.get(timeout=1) - self._approval_state = None - self._approval_deadline = 0 - self._paint_now() - _outcome_labels = { - "once": "allowed once", - "session": "allowed for session", - "always": "added to allowlist", - "deny": "denied", - } - self._persist_prompt_summary( - "⚠", "Approval", command, - _outcome_labels.get(result, str(result)), - ) - return result - except queue.Empty: - remaining = self._approval_deadline - _time.monotonic() - if remaining <= 0: - break - now = _time.monotonic() - if now - _last_countdown_refresh >= 1.0: - _last_countdown_refresh = now - self._paint_now() - + result = self._poll_modal_queue(response_queue, "_approval_deadline") self._approval_state = None self._approval_deadline = 0 self._paint_now() - _cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}") + if result is _TIMED_OUT: + _cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}") + self._persist_prompt_summary("⚠", "Approval", command, "timed out (no response)") + return "timeout" self._persist_prompt_summary( - "⚠", "Approval", command, "timed out (no response)", + "⚠", "Approval", command, _APPROVAL_OUTCOME_LABELS.get(result, str(result)), ) - return "timeout" + return result def _approval_choices(self, command: str, *, allow_permanent: bool = True, allow_session: bool = True, smart_denied: bool = False) -> list[str]: - """Return approval choices for a dangerous command prompt.""" + """Smart-DENY overrides and re-ask-every-time gates (allow_session=False) show only + once/deny; ``allow_permanent=False`` for another reason (e.g. tirith) hides only 'always'.""" if smart_denied or not allow_session: choices = ["once", "deny"] + elif allow_permanent: + choices = ["once", "session", "always", "deny"] else: - choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"] + choices = ["once", "session", "deny"] if len(command) > 70: choices.append("view") return choices def _computer_use_approval_callback(self, action: str, args: dict, summary: str) -> str: - """Adapt the generic approval UI for the computer_use tool. - - The computer_use handler expects verdicts of the form - `approve_once` | `approve_session` | `always_approve` | `deny`. - The CLI's built-in approval UI returns `once` | `session` | `always` - | `deny`. Translate between the two. - """ - # Build a command-ish string so the existing UI renders something - # meaningful. `summary` is already a one-line human description. + """Adapt the generic approval UI (once/session/always/deny) to the computer_use verdicts + (approve_once/approve_session/always_approve/deny).""" verdict = self._approval_callback( command=f"computer_use: {summary}", description=f"Allow computer_use to perform `{action}`?", @@ -1110,14 +856,12 @@ class CLIModalMixin: state = self._approval_state if not state: return - selected = state.get("selected", 0) choices = state.get("choices") if not isinstance(choices, list): choices = [] if not (0 <= selected < len(choices)): return - chosen = choices[selected] if chosen == "view": state["show_full"] = True @@ -1126,7 +870,6 @@ class CLIModalMixin: state["selected"] = max(0, len(state["choices"]) - 1) self._invalidate() return - state["response_queue"].put(chosen) self._approval_state = None self._invalidate() @@ -1140,10 +883,7 @@ class CLIModalMixin: return try: buf = self._app.current_buffer - self._modal_input_snapshot = { - "text": buf.text, - "cursor_position": buf.cursor_position, - } + self._modal_input_snapshot = {"text": buf.text, "cursor_position": buf.cursor_position} buf.reset() except Exception: self._modal_input_snapshot = None @@ -1164,18 +904,11 @@ class CLIModalMixin: def _clear_active_overlays_for_interrupt(self) -> None: """Drain and clear every input-blocking overlay left by an interrupted agent. - approval/clarify/sudo/secret prompts each block a worker thread on a - ``response_queue.get()``. When the agent is interrupted the worker - thread is torn down, but the overlay's state dict stays set — leaving - the CLI input gated (``read_only`` condition + keypress filter) with no - thread servicing the prompt. The result is a frozen terminal until the - prompt's own timeout expires. Push a terminal value onto each queue so - any still-blocked thread unblocks cleanly, then nil the state out and - restore the user's pre-modal draft (#14026). - - Safe default per prompt: approval -> "deny", clarify/sudo/secret -> - cancel (None / empty). Each step is wrapped so a dead queue can't - prevent clearing the others. + Each prompt blocks a worker thread on ``response_queue.get()``; an interrupt tears the + thread down but leaves the state dict set, gating input with nothing servicing it (frozen + terminal until the prompt's own timeout). Push a safe terminal value onto each queue + (approval -> "deny", clarify/sudo/secret -> cancel), nil the state, restore the draft. + Each step is wrapped so a dead queue can't prevent clearing the others. """ if self._approval_state: try: @@ -1213,9 +946,7 @@ class CLIModalMixin: self._secret_state["response_queue"].put(value) self._secret_state = None self._secret_deadline = 0 - # Modal teardown — paint directly so the secret panel clears at once and - # isn't held by the _invalidate throttle/resize guard (#41098). - self._paint_now() + self._paint_now() # direct paint so the secret panel clears at once (no throttle) def _cancel_secret_capture(self) -> None: self._submit_secret_response("")