From f0d24b3c2e6b4462791989098a246106eea167ae Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 18:59:19 -0700 Subject: [PATCH] refactor(cli): split _chat_render_turn and _process_unregistered_slash into per-branch helpers --- cli.py | 594 ++++++++++++++++++++++++++++----------------------------- 1 file changed, 288 insertions(+), 306 deletions(-) diff --git a/cli.py b/cli.py index ff0e5be94e..df27b397be 100644 --- a/cli.py +++ b/cli.py @@ -6283,182 +6283,151 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix expansion -> unknown-command message. Always returns True unless it re-dispatches through process_command. """ - # Check for user-defined quick commands (bypass agent loop, no LLM call) base_cmd = cmd_lower.split()[0] skill_commands = _ensure_skill_commands() skill_bundles = get_skill_bundles() quick_commands = self.config.get("quick_commands", {}) + user_args = cmd_original[len(base_cmd):].strip() if base_cmd.lstrip("/") in quick_commands: - qcmd = quick_commands[base_cmd.lstrip("/")] - if qcmd.get("type") == "exec": - import subprocess - exec_cmd = qcmd.get("command", "") - if exec_cmd: - try: - # shell=True is intentional: quick_commands are user-defined - # shell snippets from config.yaml — not agent/LLM controlled. - # Sanitize env to prevent credential leakage — - # quick commands run in the CLI process which - # has all API keys in os.environ. - from tools.environments.local import build_subprocess_env - sanitized_env = build_subprocess_env() - from hermes_cli._subprocess_compat import windows_hide_flags - result = subprocess.run( - exec_cmd, shell=True, capture_output=True, - text=True, encoding="utf-8", errors="replace", timeout=30, env=sanitized_env, - # No console flash on Windows (#56747). - creationflags=windows_hide_flags(), - ) - output = result.stdout.strip() or result.stderr.strip() - if output: - from agent.redact import redact_sensitive_text - output = redact_sensitive_text(output) - self._console_print(_rich_text_from_ansi(output)) - else: - self._console_print("[dim]Command returned no output[/]") - except subprocess.TimeoutExpired: - self._console_print("[bold red]Quick command timed out (30s)[/]") - except Exception as e: - self._console_print(f"[bold red]Quick command error: {e}[/]") - else: - self._console_print(f"[bold red]Quick command '{base_cmd}' has no command defined[/]") - elif qcmd.get("type") == "alias": - target = qcmd.get("target", "").strip() - if target: - target = target if target.startswith("/") else f"/{target}" - user_args = cmd_original[len(base_cmd):].strip() - aliased_command = f"{target} {user_args}".strip() - return self.process_command(aliased_command) - else: - self._console_print(f"[bold red]Quick command '{base_cmd}' has no target defined[/]") - else: - self._console_print(f"[bold red]Quick command '{base_cmd}' has unsupported type (supported: 'exec', 'alias')[/]") - # Check for plugin-registered slash commands - elif base_cmd.lstrip("/") in _get_plugin_cmd_handler_names(): - from hermes_cli.plugins import ( - get_plugin_command_handler, - resolve_plugin_command_result, - ) - plugin_handler = get_plugin_command_handler(base_cmd.lstrip("/")) - if plugin_handler: - user_args = cmd_original[len(base_cmd):].strip() - try: - result = resolve_plugin_command_result( - plugin_handler(user_args) - ) - if result: - _cprint(str(result)) - except Exception as e: - _cprint(f"\033[1;31mPlugin command error: {e}{_RST}") - # Skill bundles take precedence over individual skills — / - # loads multiple skills at once. Rescans cheaply when files change. + return self._run_quick_command(base_cmd, quick_commands[base_cmd.lstrip("/")], user_args) + if base_cmd.lstrip("/") in _get_plugin_cmd_handler_names(): + self._run_plugin_slash_command(base_cmd, user_args) elif base_cmd in skill_bundles: - user_instruction = cmd_original[len(base_cmd):].strip() - bundle_result = build_bundle_invocation_message( - base_cmd, user_instruction, task_id=self.session_id - ) - if bundle_result: - msg, loaded_names, missing = bundle_result - bundle_info = skill_bundles[base_cmd] - print( - f"\n⚡ Loading bundle: {bundle_info['name']} " - f"({len(loaded_names)} skills)" - ) - if missing: - ChatConsole().print( - f"[yellow]Skipped missing skills: {', '.join(missing)}[/]" - ) - if hasattr(self, '_pending_input'): - self._pending_input.put(msg) - else: - ChatConsole().print( - f"[bold red]Failed to load bundle for {base_cmd}[/]" - ) - # Check for skill slash commands (/gif-search, /axolotl, etc.) + self._run_skill_bundle_command(base_cmd, skill_bundles[base_cmd], user_args) elif base_cmd in skill_commands: - rest = cmd_original[len(base_cmd):].strip() - # Stacked slash-skill invocations: `/skill-a /skill-b do XYZ` - # loads every leading skill (up to 5), not just the first. - # Inspired by Claude Code v2.1.199. - from agent.skill_commands import ( - build_stacked_skill_invocation_message, - split_stacked_skill_commands, - ) - extra_keys, user_instruction = split_stacked_skill_commands(rest) - if extra_keys: - stacked_result = build_stacked_skill_invocation_message( - [base_cmd, *extra_keys], - user_instruction, - task_id=self.session_id, - ) - if stacked_result: - msg, loaded_names, missing = stacked_result - print( - f"\n⚡ Loading {len(loaded_names)} stacked skills: " - f"{', '.join(loaded_names)}" - ) - if missing: - ChatConsole().print( - f"[yellow]Skipped missing skills: {', '.join(missing)}[/]" - ) - if hasattr(self, '_pending_input'): - self._pending_input.put(msg) - else: - ChatConsole().print( - f"[bold red]Failed to load stacked skills for {base_cmd}[/]" - ) - return True - user_instruction = rest - msg = build_skill_invocation_message( - base_cmd, user_instruction, task_id=self.session_id - ) - if msg: - skill_name = skill_commands[base_cmd]["name"] - print(f"\n⚡ Loading skill: {skill_name}") - if hasattr(self, '_pending_input'): - self._pending_input.put(msg) - else: - ChatConsole().print(f"[bold red]Failed to load skill for {base_cmd}[/]") + self._run_skill_slash_command(base_cmd, skill_commands[base_cmd], user_args) else: - # Prefix matching: if input uniquely identifies one command, execute it. - # Matches against both built-in COMMANDS and installed skill commands so - # that execution-time resolution agrees with tab-completion. - from hermes_cli.commands import COMMANDS - typed_base = cmd_lower.split()[0] - all_known = set(COMMANDS) | set(skill_commands) | set(skill_bundles) - matches = [c for c in all_known if c.startswith(typed_base)] - if len(matches) > 1: - # Prefer an exact match (typed the full command name) - exact = [c for c in matches if c == typed_base] - if len(exact) == 1: - matches = exact + return self._expand_slash_prefix(cmd_original, cmd_lower, skill_commands, skill_bundles) + return True + + def _run_quick_command(self, base_cmd: str, qcmd: dict, user_args: str) -> bool: + """User-defined quick command (config.yaml): ``exec`` runs a shell snippet, ``alias`` re-dispatches.""" + qtype = qcmd.get("type") + if qtype == "exec": + import subprocess + exec_cmd = qcmd.get("command", "") + if not exec_cmd: + self._console_print(f"[bold red]Quick command '{base_cmd}' has no command defined[/]") + return True + try: + # shell=True is intentional: user-defined shell snippets from config.yaml, + # never agent/LLM controlled. The env is sanitized because the CLI process + # holds every API key in os.environ. + from tools.environments.local import build_subprocess_env + from hermes_cli._subprocess_compat import windows_hide_flags + result = subprocess.run( + exec_cmd, shell=True, capture_output=True, + text=True, encoding="utf-8", errors="replace", timeout=30, env=build_subprocess_env(), + # No console flash on Windows (#56747). + creationflags=windows_hide_flags(), + ) + output = result.stdout.strip() or result.stderr.strip() + if output: + from agent.redact import redact_sensitive_text + self._console_print(_rich_text_from_ansi(redact_sensitive_text(output))) else: - # Prefer the unique shortest match: - # /qui → /quit (5) wins over /quint-pipeline (15) - min_len = min(len(c) for c in matches) - shortest = [c for c in matches if len(c) == min_len] - if len(shortest) == 1: - matches = shortest - if len(matches) == 1: - # Expand the prefix to the full command name, preserving arguments. - # Guard against redispatching the same token to avoid infinite - # recursion when the expanded name still doesn't hit an exact branch - # (e.g. /config with extra args that are not yet handled above). - full_name = matches[0] - if full_name == typed_base: - # Already an exact token — no expansion possible; fall through - _cprint(f"\033[1;31mUnknown command: {cmd_lower}{_RST}") - _cprint(f"{_DIM}{_ACCENT}Type /help for available commands{_RST}") - else: - remainder = cmd_original.strip()[len(typed_base):] - full_cmd = full_name + remainder - return self.process_command(full_cmd) - elif len(matches) > 1: - _cprint(f"{_ACCENT}Ambiguous command: {cmd_lower}{_RST}") - _cprint(f"{_DIM}Did you mean: {', '.join(sorted(matches))}?{_RST}") + self._console_print("[dim]Command returned no output[/]") + except subprocess.TimeoutExpired: + self._console_print("[bold red]Quick command timed out (30s)[/]") + except Exception as e: + self._console_print(f"[bold red]Quick command error: {e}[/]") + elif qtype == "alias": + target = qcmd.get("target", "").strip() + if target: + target = target if target.startswith("/") else f"/{target}" + return self.process_command(f"{target} {user_args}".strip()) + self._console_print(f"[bold red]Quick command '{base_cmd}' has no target defined[/]") + else: + self._console_print(f"[bold red]Quick command '{base_cmd}' has unsupported type (supported: 'exec', 'alias')[/]") + return True + + def _run_plugin_slash_command(self, base_cmd: str, user_args: str) -> None: + from hermes_cli.plugins import ( + get_plugin_command_handler, + resolve_plugin_command_result, + ) + plugin_handler = get_plugin_command_handler(base_cmd.lstrip("/")) + if plugin_handler: + try: + result = resolve_plugin_command_result(plugin_handler(user_args)) + if result: + _cprint(str(result)) + except Exception as e: + _cprint(f"\033[1;31mPlugin command error: {e}{_RST}") + + def _queue_skill_message(self, msg) -> None: + if hasattr(self, '_pending_input'): + self._pending_input.put(msg) + + def _run_skill_bundle_command(self, base_cmd: str, bundle_info: dict, user_instruction: str) -> None: + """``/`` loads several skills at once (bundles win over same-named skills).""" + bundle_result = build_bundle_invocation_message( + base_cmd, user_instruction, task_id=self.session_id + ) + if not bundle_result: + ChatConsole().print(f"[bold red]Failed to load bundle for {base_cmd}[/]") + return + msg, loaded_names, missing = bundle_result + print(f"\n⚡ Loading bundle: {bundle_info['name']} ({len(loaded_names)} skills)") + if missing: + ChatConsole().print(f"[yellow]Skipped missing skills: {', '.join(missing)}[/]") + self._queue_skill_message(msg) + + def _run_skill_slash_command(self, base_cmd: str, skill_info: dict, rest: str) -> None: + """``/ ...``; stacked ``/skill-a /skill-b do XYZ`` loads every leading skill (up to 5).""" + from agent.skill_commands import ( + build_stacked_skill_invocation_message, + split_stacked_skill_commands, + ) + extra_keys, user_instruction = split_stacked_skill_commands(rest) + if extra_keys: + stacked_result = build_stacked_skill_invocation_message( + [base_cmd, *extra_keys], + user_instruction, + task_id=self.session_id, + ) + if not stacked_result: + ChatConsole().print(f"[bold red]Failed to load stacked skills for {base_cmd}[/]") + return + msg, loaded_names, missing = stacked_result + print(f"\n⚡ Loading {len(loaded_names)} stacked skills: {', '.join(loaded_names)}") + if missing: + ChatConsole().print(f"[yellow]Skipped missing skills: {', '.join(missing)}[/]") + self._queue_skill_message(msg) + return + msg = build_skill_invocation_message(base_cmd, rest, task_id=self.session_id) + if msg: + print(f"\n⚡ Loading skill: {skill_info['name']}") + self._queue_skill_message(msg) + else: + ChatConsole().print(f"[bold red]Failed to load skill for {base_cmd}[/]") + + def _expand_slash_prefix(self, cmd_original: str, cmd_lower: str, skill_commands, skill_bundles) -> bool: + """Unique-prefix expansion against built-in COMMANDS + skill commands/bundles (agrees with tab-completion).""" + from hermes_cli.commands import COMMANDS + typed_base = cmd_lower.split()[0] + all_known = set(COMMANDS) | set(skill_commands) | set(skill_bundles) + matches = [c for c in all_known if c.startswith(typed_base)] + if len(matches) > 1: + exact = [c for c in matches if c == typed_base] + if len(exact) == 1: + matches = exact else: - _cprint(f"\033[1;31mUnknown command: {cmd_lower}{_RST}") - _cprint(f"{_DIM}{_ACCENT}Type /help for available commands{_RST}") + # Unique shortest match wins: /qui → /quit (5) over /quint-pipeline (15) + min_len = min(len(c) for c in matches) + shortest = [c for c in matches if len(c) == min_len] + if len(shortest) == 1: + matches = shortest + if len(matches) == 1 and matches[0] != typed_base: + # Expand to the full name, preserving arguments. + return self.process_command(matches[0] + cmd_original.strip()[len(typed_base):]) + if len(matches) > 1: + _cprint(f"{_ACCENT}Ambiguous command: {cmd_lower}{_RST}") + _cprint(f"{_DIM}Did you mean: {', '.join(sorted(matches))}?{_RST}") + else: + # Exact token with no handler (never re-dispatch the same token: recursion), or no match. + _cprint(f"\033[1;31mUnknown command: {cmd_lower}{_RST}") + _cprint(f"{_DIM}{_ACCENT}Type /help for available commands{_RST}") return True def _owns_process_notification(self, event: dict) -> bool: @@ -7158,150 +7127,13 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix self._voice_continuous = False _cprint(f"\n{_DIM}Continuous voice mode stopped due to error.{_RST}") - # Handle interrupt - check if we were interrupted - pending_message = None - _show_interrupt_marker = False - _interrupted_this_turn = bool(turn.result and turn.result.get("interrupted")) - # Expose the flag for post-turn hooks (e.g. goal continuation) - # so they can skip themselves when the turn was user-cancelled. - self._last_turn_interrupted = _interrupted_this_turn - if _interrupted_this_turn: - pending_message = turn.result.get("interrupt_message") or interrupt_msg - # #60920: Don't append the interruption marker to response so it - # is never recorded in _OUTPUT_HISTORY by the Panel rendering - # below. The marker is printed separately with _suspend_output_history - # after the response Panel to preserve the visual while avoiding - # duplicates on terminal redraw (_recover_terminal_after_interrupt). - _show_interrupt_marker = bool(response and pending_message) - elif interrupt_msg: - # We fired agent.interrupt(interrupt_msg) but the turn result - # doesn't acknowledge it. Two ways this happens, both racy: - # 1. The agent thread had already passed its last interrupt - # check (or finished) when the interrupt landed — the turn - # completed normally and finalize_turn() never saw the flag. - # 2. The 10s post-interrupt wait above expired and we - # abandoned the daemon thread; `result` is still None. - # In both cases the user's message must NOT be dropped — - # re-queue it as the next turn (#interrupt-vacuumed-into-void). - pending_message = interrupt_msg - # If the interrupt landed after finalize_turn()'s - # clear_interrupt(), the stale flag would instantly abort the - # NEXT turn at its first loop check. Clear it now that we've - # claimed the message — but ONLY if the agent thread actually - # exited. If it's still alive (abandoned after the 10s wait), - # the flag is what makes the wedged tool eventually unwind; - # clearing it would un-signal that thread. - try: - if ( - not agent_thread.is_alive() - and self.agent - and getattr(self.agent, "_interrupt_requested", False) - ): - self.agent.clear_interrupt() - except Exception: - pass + pending_message, _show_interrupt_marker = self._chat_resolve_interrupt(turn, agent_thread, interrupt_msg, response) response_previewed = turn.result.get("response_previewed", False) if turn.result else False - # Display reasoning (thinking) box if enabled and available. - # Skip when streaming already showed reasoning live. Use the - # turn-persistent flag (_reasoning_shown_this_turn) instead of - # _reasoning_stream_started — the latter gets reset during - # intermediate turn boundaries (tool-calling loops), which caused - # the reasoning box to re-render after the final response. - _reasoning_already_shown = getattr(self, '_reasoning_shown_this_turn', False) - if self.show_reasoning and turn.result and not _reasoning_already_shown: - reasoning = turn.result.get("last_reasoning") - if reasoning: - w = self._scrollback_box_width() - r_label = " Reasoning " - r_fill = w - 2 - len(r_label) - r_top = f"{_DIM}┌─{r_label}{'─' * max(r_fill - 1, 0)}┐{_RST}" - r_bot = f"{_DIM}└{'─' * (w - 2)}┘{_RST}" - # Collapse long reasoning to the first 10 lines unless the - # user opted into full display via /reasoning full. - lines = reasoning.strip().splitlines() - if len(lines) > 10 and not getattr(self, "reasoning_full", False): - display_reasoning = "\n".join(lines[:10]) - display_reasoning += f"\n{_DIM} ... ({len(lines) - 10} more lines — /reasoning full to show){_RST}" - else: - display_reasoning = reasoning.strip() - _cprint(f"\n{r_top}\n{_DIM}{display_reasoning}{_RST}\n{r_bot}") + self._chat_print_reasoning_box(turn) - if response and not response_previewed: - # Use skin engine for label/color with fallback - try: - from hermes_cli.skin_engine import get_active_skin - _skin = get_active_skin() - label = _skin.get_branding("response_label", "⚕ Hermes") - _resp_color = _maybe_remap_for_light_mode(_skin.get_color("response_border", "#CD7F32")) - _resp_text = _maybe_remap_for_light_mode(_skin.get_color("banner_text", "#FFF8DC")) - except Exception: - label = "⚕ Hermes" - _resp_color = _maybe_remap_for_light_mode("#CD7F32") - _resp_text = _maybe_remap_for_light_mode("#FFF8DC") - - is_error_response = turn.result and (turn.result.get("failed") or turn.result.get("partial")) - already_streamed = self._stream_started and self._stream_box_opened and not is_error_response - if turn.use_streaming_tts and turn.box_opened and not is_error_response: - # Text was already printed sentence-by-sentence; just close the box - w = self._scrollback_box_width() - _cprint(f"\n{_ACCENT}╰{'─' * (w - 2)}╯{_RST}") - elif already_streamed: - # Response was already streamed token-by-token with box framing; - # _flush_stream() already closed the box. Skip Rich Panel. - # A transform hook runs after streaming. Show a suffix for - # append-only changes, or the complete replacement otherwise. - _post_stream_text = _post_stream_transform_output(response, turn.result) - if _post_stream_text.strip(): - _cprint(_post_stream_text) - else: - _chat_console = ChatConsole() - _chat_console.print(Panel( - _render_final_assistant_content(response, mode=self.final_response_markdown), - title=f"[{_resp_color} bold]{label}[/]", - title_align="left", - border_style=_resp_color, - style=_resp_text, - box=rich_box.HORIZONTALS, - padding=(1, 0), - width=self._scrollback_box_width(), - )) - - # Durable, provider-agnostic billing CTA below the response. The - # response panel carries the full guidance; this pins the single - # action to take (Nous → /topup, other providers → their billing - # page) so it stays visible instead of scrolling away as prose. - if turn.result and turn.result.get("failure_reason") == "billing": - _bb = turn.result.get("billing_block") or {} - _prov_label = _bb.get("provider_label") or "your provider" - if _bb.get("is_nous"): - _cta_lines = [ - "Run [bold]/topup[/] to add credits, or " - "[bold]/subscription[/] to change plan.", - ] - else: - _url = _bb.get("billing_url") - _cta_lines = [ - f"Add credits with {_prov_label}" - + (f": [bold]{_url}[/]" if _url else ".") - ] - _cta_lines.append( - "Or switch providers with " - "[bold]/model --provider [/]." - ) - try: - ChatConsole().print(Panel( - "\n".join(_cta_lines), - title="[#CD7F32 bold]⚡ Out of credits[/]", - title_align="left", - border_style="#CD7F32", - box=rich_box.HORIZONTALS, - padding=(1, 4), - width=self._scrollback_box_width(), - )) - except Exception: - pass + self._chat_print_response_panel(turn, response, response_previewed) # #60920: Print interruption marker with history suppressed so it # is never recorded in _OUTPUT_HISTORY. The marker was previously @@ -7377,6 +7209,156 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix return response + def _chat_resolve_interrupt(self, turn, agent_thread, interrupt_msg, response): + """Decide the re-queued interrupt message and whether to print the interrupt marker; clears a stale agent interrupt flag.""" + # Handle interrupt - check if we were interrupted + pending_message = None + _show_interrupt_marker = False + _interrupted_this_turn = bool(turn.result and turn.result.get("interrupted")) + # Expose the flag for post-turn hooks (e.g. goal continuation) + # so they can skip themselves when the turn was user-cancelled. + self._last_turn_interrupted = _interrupted_this_turn + if _interrupted_this_turn: + pending_message = turn.result.get("interrupt_message") or interrupt_msg + # #60920: Don't append the interruption marker to response so it + # is never recorded in _OUTPUT_HISTORY by the Panel rendering + # below. The marker is printed separately with _suspend_output_history + # after the response Panel to preserve the visual while avoiding + # duplicates on terminal redraw (_recover_terminal_after_interrupt). + _show_interrupt_marker = bool(response and pending_message) + elif interrupt_msg: + # We fired agent.interrupt(interrupt_msg) but the turn result + # doesn't acknowledge it. Two ways this happens, both racy: + # 1. The agent thread had already passed its last interrupt + # check (or finished) when the interrupt landed — the turn + # completed normally and finalize_turn() never saw the flag. + # 2. The 10s post-interrupt wait above expired and we + # abandoned the daemon thread; `result` is still None. + # In both cases the user's message must NOT be dropped — + # re-queue it as the next turn (#interrupt-vacuumed-into-void). + pending_message = interrupt_msg + # If the interrupt landed after finalize_turn()'s + # clear_interrupt(), the stale flag would instantly abort the + # NEXT turn at its first loop check. Clear it now that we've + # claimed the message — but ONLY if the agent thread actually + # exited. If it's still alive (abandoned after the 10s wait), + # the flag is what makes the wedged tool eventually unwind; + # clearing it would un-signal that thread. + try: + if ( + not agent_thread.is_alive() + and self.agent + and getattr(self.agent, "_interrupt_requested", False) + ): + self.agent.clear_interrupt() + except Exception: + pass + return pending_message, _show_interrupt_marker + + def _chat_print_reasoning_box(self, turn): + """Collapsed reasoning box when show_reasoning is on and streaming did not already show it this turn.""" + # Display reasoning (thinking) box if enabled and available. + # Skip when streaming already showed reasoning live. Use the + # turn-persistent flag (_reasoning_shown_this_turn) instead of + # _reasoning_stream_started — the latter gets reset during + # intermediate turn boundaries (tool-calling loops), which caused + # the reasoning box to re-render after the final response. + _reasoning_already_shown = getattr(self, '_reasoning_shown_this_turn', False) + if self.show_reasoning and turn.result and not _reasoning_already_shown: + reasoning = turn.result.get("last_reasoning") + if reasoning: + w = self._scrollback_box_width() + r_label = " Reasoning " + r_fill = w - 2 - len(r_label) + r_top = f"{_DIM}┌─{r_label}{'─' * max(r_fill - 1, 0)}┐{_RST}" + r_bot = f"{_DIM}└{'─' * (w - 2)}┘{_RST}" + # Collapse long reasoning to the first 10 lines unless the + # user opted into full display via /reasoning full. + lines = reasoning.strip().splitlines() + if len(lines) > 10 and not getattr(self, "reasoning_full", False): + display_reasoning = "\n".join(lines[:10]) + display_reasoning += f"\n{_DIM} ... ({len(lines) - 10} more lines — /reasoning full to show){_RST}" + else: + display_reasoning = reasoning.strip() + _cprint(f"\n{r_top}\n{_DIM}{display_reasoning}{_RST}\n{r_bot}") + + def _chat_print_response_panel(self, turn, response, response_previewed): + """Response box: close the TTS-drawn box, print the post-stream transform, or render the Rich Panel; then the billing CTA.""" + if response and not response_previewed: + # Use skin engine for label/color with fallback + try: + from hermes_cli.skin_engine import get_active_skin + _skin = get_active_skin() + label = _skin.get_branding("response_label", "⚕ Hermes") + _resp_color = _maybe_remap_for_light_mode(_skin.get_color("response_border", "#CD7F32")) + _resp_text = _maybe_remap_for_light_mode(_skin.get_color("banner_text", "#FFF8DC")) + except Exception: + label = "⚕ Hermes" + _resp_color = _maybe_remap_for_light_mode("#CD7F32") + _resp_text = _maybe_remap_for_light_mode("#FFF8DC") + + is_error_response = turn.result and (turn.result.get("failed") or turn.result.get("partial")) + already_streamed = self._stream_started and self._stream_box_opened and not is_error_response + if turn.use_streaming_tts and turn.box_opened and not is_error_response: + # Text was already printed sentence-by-sentence; just close the box + w = self._scrollback_box_width() + _cprint(f"\n{_ACCENT}╰{'─' * (w - 2)}╯{_RST}") + elif already_streamed: + # Response was already streamed token-by-token with box framing; + # _flush_stream() already closed the box. Skip Rich Panel. + # A transform hook runs after streaming. Show a suffix for + # append-only changes, or the complete replacement otherwise. + _post_stream_text = _post_stream_transform_output(response, turn.result) + if _post_stream_text.strip(): + _cprint(_post_stream_text) + else: + _chat_console = ChatConsole() + _chat_console.print(Panel( + _render_final_assistant_content(response, mode=self.final_response_markdown), + title=f"[{_resp_color} bold]{label}[/]", + title_align="left", + border_style=_resp_color, + style=_resp_text, + box=rich_box.HORIZONTALS, + padding=(1, 0), + width=self._scrollback_box_width(), + )) + + # Durable, provider-agnostic billing CTA below the response. The + # response panel carries the full guidance; this pins the single + # action to take (Nous → /topup, other providers → their billing + # page) so it stays visible instead of scrolling away as prose. + if turn.result and turn.result.get("failure_reason") == "billing": + _bb = turn.result.get("billing_block") or {} + _prov_label = _bb.get("provider_label") or "your provider" + if _bb.get("is_nous"): + _cta_lines = [ + "Run [bold]/topup[/] to add credits, or " + "[bold]/subscription[/] to change plan.", + ] + else: + _url = _bb.get("billing_url") + _cta_lines = [ + f"Add credits with {_prov_label}" + + (f": [bold]{_url}[/]" if _url else ".") + ] + _cta_lines.append( + "Or switch providers with " + "[bold]/model --provider [/]." + ) + try: + ChatConsole().print(Panel( + "\n".join(_cta_lines), + title="[#CD7F32 bold]⚡ Out of credits[/]", + title_align="left", + border_style="#CD7F32", + box=rich_box.HORIZONTALS, + padding=(1, 4), + width=self._scrollback_box_width(), + )) + except Exception: + pass + def _tui_process_loop(self): while not self._should_exit: try: