From 0a99ce8d5b6cbb12e8c4920aa5ffabe00ed70e51 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:42:46 -0700 Subject: [PATCH] refactor(hermes_cli): AST-neutral bracket packing in info/modal mixins and callbacks --- hermes_cli/callbacks.py | 9 ++-- hermes_cli/cli_info_mixin.py | 81 ++++++++++++----------------------- hermes_cli/cli_modal_mixin.py | 69 +++++++++-------------------- 3 files changed, 50 insertions(+), 109 deletions(-) diff --git a/hermes_cli/callbacks.py b/hermes_cli/callbacks.py index b8a0c897ab..d7fa16fa2f 100644 --- a/hermes_cli/callbacks.py +++ b/hermes_cli/callbacks.py @@ -28,8 +28,7 @@ def _clear_secret_input(cli) -> None: def _skipped(var_name: str, reason: str, message: str) -> dict: return { "success": True, "reason": reason, "stored_as": var_name, - "validated": False, "skipped": True, "message": message, - } + "validated": False, "skipped": True, "message": message} def _secret_result(var_name: str, value: str) -> dict: @@ -42,8 +41,7 @@ def _secret_result(var_name: str, value: str) -> dict: return { **stored, "skipped": False, - "message": "Secret stored securely. The secret value was not exposed to the model.", - } + "message": "Secret stored securely. The secret value was not exposed to the model."} def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict: @@ -68,8 +66,7 @@ def prompt_for_secret(cli, var_name: str, prompt: str, metadata=None) -> dict: "var_name": var_name, "prompt": prompt, "metadata": metadata or {}, - "response_queue": response_queue, - } + "response_queue": response_queue} cli._secret_deadline = _time.monotonic() + 120 if hasattr(cli, "_ring_bell"): cli._ring_bell(prompt=True, context=f"secret needed ({var_name})") diff --git a/hermes_cli/cli_info_mixin.py b/hermes_cli/cli_info_mixin.py index 28659f7bf9..7e8d35f781 100644 --- a/hermes_cli/cli_info_mixin.py +++ b/hermes_cli/cli_info_mixin.py @@ -27,14 +27,12 @@ _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"), -] + ("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)." -) + "high-reasoning models).") def _ascii_box(title: str, width: int) -> None: @@ -82,8 +80,7 @@ class CLIInfoMixin: # 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) try: snapshot = load_banner_snapshot(self.enabled_toolsets) except Exception: @@ -92,8 +89,7 @@ class CLIInfoMixin: banner_kw = dict( console=self.console, model=self.model, cwd=cwd, enabled_toolsets=self.enabled_toolsets, session_id=self.session_id, - context_length=ctx_len, provider=self.provider, - ) + context_length=ctx_len, provider=self.provider) if snapshot is not None: self._defer_tool_warnings = True @@ -103,15 +99,13 @@ class CLIInfoMixin: get_toolset_for_tool=lambda name: toolset_map.get(name), availability=snapshot["availability"], skills_by_category=snapshot.get("skills_by_category"), - **banner_kw, - ) + **banner_kw) def _refresh_banner_snapshot() -> None: try: from model_tools import get_toolset_for_tool tools = get_tool_definitions( - enabled_toolsets=self.enabled_toolsets, quiet_mode=True - ) + enabled_toolsets=self.enabled_toolsets, quiet_mode=True) availability = compute_toolset_availability(self.enabled_toolsets) tmap = _toolset_map(tools, availability, get_toolset_for_tool) save_banner_snapshot(tools, self.enabled_toolsets, availability, tmap) @@ -141,8 +135,7 @@ class CLIInfoMixin: threading.Thread( target=self._show_tool_availability_warnings, name="tool-availability-warnings", - daemon=True, - ).start() + daemon=True).start() else: self._show_tool_availability_warnings() @@ -152,8 +145,7 @@ class CLIInfoMixin: self._console_print() self._console_print( f"[yellow]⚠️ Context length is only {ctx_len:,} tokens — " - f"this is likely too low for agent use with tools.[/]" - ) + f"this is likely too low for agent use with tools.[/]") self._console_print( f"[dim] Hermes needs at least {MINIMUM_CONTEXT_LENGTH:,} tokens. Tool schemas + system prompt use a large fixed prefix.[/]" ) @@ -181,20 +173,17 @@ class CLIInfoMixin: self._console_print() self._console_print( "[bold yellow]⚠ Nous Research Hermes 3 & 4 models are NOT agentic and are not " - "designed for use with Hermes Agent.[/]" - ) + "designed for use with Hermes Agent.[/]") self._console_print( "[dim] They lack tool-calling capabilities required for agent workflows. " - "Consider using an agentic model (Claude, GPT, Gemini, DeepSeek, etc.).[/]" - ) + "Consider using an agentic model (Claude, GPT, Gemini, DeepSeek, etc.).[/]") self._console_print("[dim] Switch with: /model sonnet or /model gpt5[/]") # 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) @@ -206,8 +195,7 @@ class CLIInfoMixin: _root, _n = _untrusted self._console_print( f"[yellow]◆ {_n} project skill(s) found in {_root} but not " - f"loaded — run `hermes skills trust` to enable them.[/]" - ) + f"loaded — run `hermes skills trust` to enable them.[/]") except Exception: logger.debug("project skills banner notice failed", exc_info=True) @@ -231,8 +219,7 @@ class CLIInfoMixin: 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, - ) + _termux_example_image_path, get_skill_bundles) from hermes_cli.commands import COMMANDS_BY_CATEGORY, HELP_SESSION_SUBGROUPS arg = (arg or "").strip() @@ -240,8 +227,7 @@ class CLIInfoMixin: def _row(cmd: str, desc: str, width: int = 15) -> None: ChatConsole().print( - f" [bold {_accent_hex()}]{cmd:<{width}}[/] [dim]-[/] {_escape(desc)}" - ) + 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. @@ -300,8 +286,7 @@ class CLIInfoMixin: if query: matched_skills = [ (cmd, info) for cmd, info in sorted(skill_commands.items()) - if query in cmd.lower() or query in (info.get("description", "").lower()) - ] + if query in cmd.lower() or query in (info.get("description", "").lower())] if matched_skills: _cprint(f"\n ⚡ {_BOLD}Skill Commands{_RST} (matching '{arg}'):") for cmd, info in matched_skills: @@ -309,8 +294,7 @@ class CLIInfoMixin: elif skill_commands: _cprint( f"\n ⚡ {_BOLD}Skill Commands{_RST}: {len(skill_commands)} installed " - f"— {_DIM}/help skills{_RST} to list them" - ) + f"— {_DIM}/help skills{_RST} to list them") _bundles_now = get_skill_bundles() if _bundles_now and not query: @@ -320,8 +304,7 @@ class CLIInfoMixin: desc = info.get("description") or f"Load {skill_count} skills" ChatConsole().print( f" [bold {_accent_hex()}]{cmd:<22}[/] [dim]-[/] " - f"{_escape(desc)} [dim]({skill_count} skills)[/]" - ) + f"{_escape(desc)} [dim]({skill_count} skills)[/]") quick_commands = self.config.get("quick_commands", {}) if quick_commands and not query: @@ -437,8 +420,7 @@ class CLIInfoMixin: return self._busy_inline_command(text, has_images, ("steer",)) def _should_handle_background_command_inline( - self, text: str, has_images: bool = False - ) -> bool: + self, text: str, has_images: bool = False) -> bool: """Return True when /bg or /btw should be dispatched while the agent runs (their ``CommandDef`` entries declare ``busy_policy="dispatch"``; the classic CLI honours it here).""" return self._busy_inline_command(text, has_images, ("bg", "btw")) @@ -455,8 +437,7 @@ class CLIInfoMixin: 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, - ) + parse_bang_command, resolve_bang_cwd, run_bang_command) if not is_bang_command(text): return False @@ -473,16 +454,14 @@ class CLIInfoMixin: approval = check_bang_approval(command) if not approval.get("approved"): message = approval.get("message") or ( - f"Command denied: {approval.get('description', 'flagged as dangerous')}" - ) + f"Command denied: {approval.get('description', 'flagged as dangerous')}") self._console_print(f"[bold red]{_escape(str(message))}[/]") return True exit_code = run_bang_command( command, cwd=resolve_bang_cwd(getattr(self, "session_id", None)), - writer=lambda line: self._console_print(_rich_text_from_ansi(line)), - ) + writer=lambda line: self._console_print(_rich_text_from_ansi(line))) if exit_code: self._console_print(f"[dim]! exited {exit_code}[/]") return True @@ -506,8 +485,7 @@ class CLIInfoMixin: Platform.TELEGRAM: ("Telegram", "TELEGRAM_BOT_TOKEN"), Platform.DISCORD: ("Discord", "DISCORD_BOT_TOKEN"), Platform.SLACK: ("Slack", "SLACK_BOT_TOKEN"), - Platform.WHATSAPP: ("WhatsApp", "WHATSAPP_ENABLED"), - } + Platform.WHATSAPP: ("WhatsApp", "WHATSAPP_ENABLED")} for platform, (name, env_var) in platform_status.items(): pconfig = config.platforms.get(platform) if pconfig and pconfig.enabled: @@ -627,10 +605,8 @@ class CLIInfoMixin: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as _pool: try: result = _pool.submit( - redeem_codex_reset_credit, - base_url=self._agent_or_self("base_url"), - api_key=self._agent_or_self("api_key"), - force=force, + redeem_codex_reset_credit, base_url=self._agent_or_self("base_url"), + api_key=self._agent_or_self("api_key"), force=force, ).result(timeout=45.0) except concurrent.futures.TimeoutError: print(" ❌ Timed out talking to the Codex backend — try again shortly.") @@ -650,8 +626,7 @@ class CLIInfoMixin: from agent.context_breakdown import ( compute_context_details, compute_session_context_breakdown, - render_context_breakdown_lines, - ) + render_context_breakdown_lines) try: payload = compute_session_context_breakdown(self.agent, self.conversation_history) except Exception as e: @@ -737,8 +712,7 @@ class CLIInfoMixin: with concurrent.futures.ThreadPoolExecutor(max_workers=1) as _pool: try: account_snapshot = _pool.submit( - fetch_account_usage, provider, - base_url=self._agent_or_self("base_url"), + fetch_account_usage, provider, base_url=self._agent_or_self("base_url"), api_key=self._agent_or_self("api_key"), ).result(timeout=10.0) except (concurrent.futures.TimeoutError, Exception): @@ -867,8 +841,7 @@ class CLIInfoMixin: choices=_RELOAD_MCP_CHOICES, unchanged="MCP tools unchanged.", always_msg="🔒 Future /reload-mcp calls will run without confirmation.", - once_verb="reloading", - ) + once_verb="reloading") if choice is None: return with self._busy_command(self._slow_command_status(cmd_original)): diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index 80130366e5..14e34458bd 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -22,20 +22,17 @@ _TIMED_OUT = object() # sentinel returned by _poll_modal_queue when the deadlin _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", -} + "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", -} + "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." -) + "Use your best judgement to make the choice and proceed.") def _approval_gate_on(key: str) -> bool: @@ -284,12 +281,7 @@ class CLIModalMixin: paint() def _prompt_text_input_modal( - self, - *, - title: str, - detail: str, - choices: list[tuple[str, str, str]], - timeout: float = 120, + self, *, title: str, detail: str, choices: list[tuple[str, str, str]], timeout: float = 120 ) -> str | None: """Slash-command confirmation through the prompt_toolkit composer instead of raw input(). @@ -329,8 +321,7 @@ class CLIModalMixin: "detail": detail, "choices": choices, "selected": 0, - "response_queue": response_queue, - } + "response_queue": response_queue} self._slash_confirm_deadline = _time.monotonic() + timeout self._invalidate() @@ -362,8 +353,7 @@ class CLIModalMixin: return _stdin_fallback() try: result = self._poll_modal_queue( - response_queue, "_slash_confirm_deadline", refresh=5.0, paint=self._invalidate, - ) + 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 @@ -382,10 +372,7 @@ class CLIModalMixin: self._invalidate() def _normalize_slash_confirm_choice( - self, - raw: str | None, - choices: list[tuple[str, str, str]], - ) -> str | None: + self, raw: str | None, choices: list[tuple[str, str, str]]) -> str | None: if raw is None: return None choice_raw = raw.strip().lower() @@ -430,8 +417,7 @@ class CLIModalMixin: "entries": self._build_command_palette_entries(), "filter": "", "selected": 0, - "_scroll_offset": 0, - } + "_scroll_offset": 0} self._invalidate(min_interval=0.0) def _close_command_palette(self) -> None: @@ -516,11 +502,7 @@ class CLIModalMixin: return " ".join(kept), len(kept) != len(tokens) def _confirm_destructive_slash( - self, - command: str, - detail: str, - cmd_original: Optional[str] = None, - ) -> Optional[str]: + self, command: str, detail: str, cmd_original: Optional[str] = None) -> Optional[str]: """Confirm a destructive session slash command (``/clear``, ``/new``/``/reset``, ``/undo``). Returns ``"once"``, ``"always"`` (also persists ``approvals.destructive_slash_confirm: @@ -538,12 +520,10 @@ class CLIModalMixin: choices=[ ("once", "Approve Once", "proceed this time only"), ("always", "Always Approve", "proceed and silence this prompt permanently"), - ("cancel", "Cancel", "keep current conversation"), - ], + ("cancel", "Cancel", "keep current conversation")], unchanged="Conversation unchanged.", always_msg="🔒 Future /clear, /new, /reset, and /undo will run without confirmation.", - once_verb="proceeding", - ) + once_verb="proceeding") 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. @@ -567,8 +547,7 @@ class CLIModalMixin: context or ("input needed" if prompt else "turn complete"), prompt=prompt, session_id=getattr(self, "session_id", "") or "", - detail=detail, - ) + detail=detail) except Exception: pass @@ -602,8 +581,7 @@ class CLIModalMixin: "selected": 0, "multi_select": effective_multi, "selected_indices": set() if effective_multi else None, - "response_queue": response_queue, - } + "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 @@ -704,8 +682,7 @@ class CLIModalMixin: self._clarify_batch_lock( state, json.dumps(selected_choices, ensure_ascii=False), - meta={"kind": "multi", "choices": selected_choices, "other_text": ""}, - ) + meta={"kind": "multi", "choices": selected_choices, "other_text": ""}) return if selected < len(choices): self._clarify_batch_lock(state, choices[selected], meta={"kind": "choice"}) @@ -734,8 +711,7 @@ class CLIModalMixin: "choices": [], "selected": 0, "multi_select": False, - "selected_indices": None, - } + "selected_indices": None} self._clarify_state = state self._clarify_batch_set_active(state, 0) self._clarify_deadline = None if timeout <= 0 else _time.monotonic() + timeout @@ -799,11 +775,9 @@ class CLIModalMixin: command, allow_permanent=allow_permanent, allow_session=allow_session, - smart_denied=smart_denied, - ), + smart_denied=smart_denied), "selected": 0, - "response_queue": response_queue, - } + "response_queue": response_queue} self._approval_deadline = _time.monotonic() + timeout self._ring_bell(prompt=True, context="approval", detail=command) self._paint_now() @@ -817,8 +791,7 @@ class CLIModalMixin: self._persist_prompt_summary("⚠", "Approval", command, "timed out (no response)") return "timeout" self._persist_prompt_summary( - "⚠", "Approval", command, _APPROVAL_OUTCOME_LABELS.get(result, str(result)), - ) + "⚠", "Approval", command, _APPROVAL_OUTCOME_LABELS.get(result, str(result))) return result def _approval_choices(self, command: str, *, allow_permanent: bool = True, @@ -841,8 +814,7 @@ class CLIModalMixin: (approve_once/approve_session/always_approve/deny).""" verdict = self._approval_callback( command=f"computer_use: {summary}", - description=f"Allow computer_use to perform `{action}`?", - ) + description=f"Allow computer_use to perform `{action}`?") return { "once": "approve_once", "session": "approve_session", @@ -919,8 +891,7 @@ class CLIModalMixin: if self._clarify_state: try: self._clarify_state["response_queue"].put( - "The user cancelled. Use your best judgement to proceed." - ) + "The user cancelled. Use your best judgement to proceed.") except Exception: pass self._clarify_state = None