From da6991f8f5a8b7870bd6375e11d3236dd9891822 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:27:53 -0700 Subject: [PATCH 1/2] refactor(hermes_cli): extract recap-entry collection, compact docstrings, squeeze intra-function blanks across g6 files --- hermes_cli/claw.py | 15 +- hermes_cli/cli_agent_setup_mixin.py | 264 ++++++++----------- hermes_cli/cli_loops_mixin.py | 82 ++---- hermes_cli/cli_terminal_mixin.py | 139 ++++------ hermes_cli/clipboard.py | 7 +- hermes_cli/codex_runtime_plugin_migration.py | 72 ++--- 6 files changed, 201 insertions(+), 378 deletions(-) diff --git a/hermes_cli/claw.py b/hermes_cli/claw.py index 089348e012..c1d6c0c1a6 100644 --- a/hermes_cli/claw.py +++ b/hermes_cli/claw.py @@ -115,8 +115,7 @@ def _detect_openclaw_processes() -> list[str]: ps_cmd = ( 'Get-CimInstance Win32_Process -Filter "Name = \'node.exe\'" | ' 'Where-Object { $_.CommandLine -match "openclaw|clawd" } | ' - 'Select-Object -First 1 ProcessId' - ) + 'Select-Object -First 1 ProcessId') result = bounded_probe_run(["powershell", "-NoProfile", "-Command", ps_cmd], timeout=5) pid = (result.stdout or "").strip() if result is not None else "" if pid: @@ -124,7 +123,6 @@ def _detect_openclaw_processes() -> list[str]: except Exception: pass return found - result = _posix_probe(["systemctl", "--user", "is-active", "openclaw-gateway.service"], 5) if result is not None and result.stdout.strip() == "active": found.append("systemd service: openclaw-gateway.service") @@ -160,7 +158,6 @@ def _warn_if_openclaw_running(auto_yes: bool) -> None: def _warn_if_gateway_running(auto_yes: bool) -> None: """Warn if a Hermes gateway has connected platforms (token conflicts, e.g. Telegram 409).""" from gateway.status import get_running_pid, read_runtime_status - platforms = ((read_runtime_status() or {}).get("platforms") or {}) if get_running_pid() else {} connected = [name for name, info in platforms.items() if isinstance(info, dict) and info.get("state") == "connected"] @@ -299,8 +296,7 @@ def _load_migrator(script_path: Path, opts: SimpleNamespace) -> Optional[Callabl source_root=opts.source_dir.resolve(), target_root=opts.hermes_home.resolve(), execute=execute, workspace_target=ws_target, overwrite=opts.overwrite, migrate_secrets=opts.migrate_secrets, output_dir=None, selected_options=selected, - preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict, - ).migrate() + preset_name=opts.preset, skill_conflict_mode=opts.skill_conflict).migrate() def _preview_migration(run_migrator: Callable[[bool], dict], opts: SimpleNamespace) -> bool: @@ -387,8 +383,7 @@ def _cmd_cleanup(args): "Stop OpenClaw first: systemctl --user stop openclaw-gateway.service"), "Proceed anyway?", declined="Aborted. Stop OpenClaw first, then re-run: hermes claw cleanup", - non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",), - ): + non_tty=("Non-interactive session — aborting. Stop OpenClaw and re-run.",)): return total_archived = 0 for source_dir in dirs_to_check: @@ -406,7 +401,6 @@ def _cmd_cleanup(args): except OSError as e: print_error(f"Could not archive: {e}") print_info(f"Try manually: mv {source_dir} {source_dir}.pre-migration") - print() n = len(dirs_to_check) if dry_run else total_archived word = "directory" if n == 1 else "directories" @@ -495,8 +489,7 @@ def _print_migration_report(report: dict, dry_run: bool): print() for line in ( " ⚠ API keys were NOT migrated (secrets migration is disabled by default).", - " Your OPENROUTER_API_KEY and other provider keys must be added manually.", - ): + " Your OPENROUTER_API_KEY and other provider keys must be added manually."): print(color(line, Colors.YELLOW)) _info("", "To migrate API keys, re-run with:", " hermes claw migrate --migrate-secrets", "", "Or add your key manually:", diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index 582b1e4b5e..afda620ee5 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -40,16 +40,14 @@ def _current_runtime(cli) -> dict: "api_mode": cli.api_mode, "command": cli.acp_command, "args": list(cli.acp_args or []), - "credential_pool": getattr(cli, "_credential_pool", None), - } + "credential_pool": getattr(cli, "_credential_pool", None)} def _route_signature(model, runtime: dict) -> tuple: """Hashable identity of (model, routing) used to detect when the agent must be rebuilt.""" return ( model, runtime.get("provider"), runtime.get("requested_provider"), runtime.get("base_url"), - runtime.get("api_mode"), runtime.get("command"), tuple(runtime.get("args") or ()), - ) + runtime.get("api_mode"), runtime.get("command"), tuple(runtime.get("args") or ())) def _keyless_custom_base(base_url) -> bool: @@ -58,8 +56,7 @@ def _keyless_custom_base(base_url) -> bool: return bool( isinstance(base_url, str) and base_url - and not base_url_host_matches(base_url, "openrouter.ai") - ) + and not base_url_host_matches(base_url, "openrouter.ai")) def _compression_descendant(session_db, session_id): @@ -78,8 +75,7 @@ def _user_display_text(content) -> str: return " ".join( part.get("text", "") if part.get("type") == "text" else "[image]" for part in content - if isinstance(part, dict) and part.get("type") in ("text", "image_url") - ) + if isinstance(part, dict) and part.get("type") in ("text", "image_url")) return "" if content is None else str(content) @@ -103,6 +99,60 @@ _RESUME_EVENT_TEXT = { "auto_continue": "resumed interrupted turn", } +def _collect_resume_entries(display_history, disp: dict, clean_assistant): + """Displayable ``(role, text)`` recap entries from stored history, truncated per the + ``display.resume_*`` config; system and tool-result rows are skipped. Returns + ``(entries, index of last assistant entry, its un-truncated text)``. + + Stored history is untrusted for display: text is sanitized so replay can't clear the + screen, retitle the window or restyle the panel. Pure-reasoning assistant rows with no + visible output are skipped, as are tool-call-only rows when ``resume_skip_tool_only``. + """ + from tools.ansi_strip import sanitize_display_text as _sanitize_display_text + max_user_len = int(disp.get("resume_max_user_chars", 300)) + max_asst_len = int(disp.get("resume_max_assistant_chars", 200)) + max_asst_lines = int(disp.get("resume_max_assistant_lines", 3)) + skip_tool_only = disp.get("resume_skip_tool_only", True) + entries: list = [] + last_asst_idx = None + last_asst_full = None + for msg in display_history: + role = msg.get("role", "") + display_kind = msg.get("display_kind") + content = msg.get("content") + tool_calls = msg.get("tool_calls") or [] + if display_kind == "hidden": + continue + if display_kind in _RESUME_EVENT_TEXT: + entries.append(("event", _RESUME_EVENT_TEXT[display_kind])) + continue + if role == "user": + text = _sanitize_display_text(_user_display_text(content)) + if len(text) > max_user_len: + text = text[:max_user_len] + "..." + entries.append(("user", text)) + elif role == "assistant": + text = clean_assistant("" if content is None else str(content)) + parts, full_parts = [], [] + if text: + full_parts.append(text) + lines = text.splitlines() + if len(lines) > max_asst_lines: + text = "\n".join(lines[:max_asst_lines]) + " ..." + if len(text) > max_asst_len: + text = text[:max_asst_len] + "..." + parts.append(text) + if tool_calls: + parts.append(_tool_calls_summary(tool_calls)) + full_parts.append(parts[-1]) + if not text and (skip_tool_only or not tool_calls): + continue + entries.append(("assistant", " ".join(parts))) + last_asst_idx = len(entries) - 1 + last_asst_full = " ".join(full_parts) + return entries, last_asst_idx, last_asst_full + + # (skin key, fallback) for recap panel colors: body text, session label, border, assistant label. _RESUME_SKIN_COLORS = ( ("banner_text", "#FFF8DC"), ("session_label", "#DAA520"), ("session_border", "#8B8682"), @@ -110,6 +160,16 @@ _RESUME_SKIN_COLORS = ( ) +def _resume_panel_colors() -> tuple: + """Active-skin colors for ``_RESUME_SKIN_COLORS`` (fallbacks when no skin loads).""" + try: + from hermes_cli.skin_engine import get_active_skin + _skin = get_active_skin() + return tuple(_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS) + except Exception: + return tuple(default for _, default in _RESUME_SKIN_COLORS) + + class CLIAgentSetupMixin: """Agent construction + session-resume display methods for ``HermesCLI``.""" @@ -118,36 +178,28 @@ class CLIAgentSetupMixin: refresh are picked up without restarting the CLI. False on auth failure.""" from cli import ChatConsole, logger from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error - _primary_exc = None runtime = None try: runtime = resolve_runtime_provider( requested=self.requested_provider, explicit_api_key=self._explicit_api_key, - explicit_base_url=self._explicit_base_url, - ) + explicit_base_url=self._explicit_base_url) except Exception as exc: _primary_exc = exc - - if runtime is None and _primary_exc is not None: - from hermes_cli.auth import AuthError - if isinstance(_primary_exc, AuthError): - runtime = self._resolve_fallback_runtime(_primary_exc) - if runtime is not None: - _primary_exc = None - + if _primary_exc is not None: + runtime = self._resolve_fallback_runtime(_primary_exc) + if runtime is not None: + _primary_exc = None if runtime is None: message = format_runtime_provider_error(_primary_exc) if _primary_exc else "Provider resolution failed." ChatConsole().print(f"[bold red]{message}[/]") return False - api_key = runtime.get("api_key") base_url = runtime.get("base_url") resolved_provider = runtime.get("provider", "openrouter") resolved_routing = ( resolved_provider, runtime.get("api_mode", self.api_mode), runtime.get("command"), - list(runtime.get("args") or []), - ) + list(runtime.get("args") or [])) # A callable api_key is a bearer-token provider (Azure Entra ID): the OpenAI SDK # invokes it per request, so skip string validation / placeholder substitution. _is_callable_provider = callable(api_key) and not isinstance(api_key, str) @@ -158,8 +210,7 @@ class CLIAgentSetupMixin: logger.debug( "No API key for custom endpoint %s (source=%s), " "using placeholder — local servers typically ignore auth", - base_url, runtime.get("source", ""), - ) + base_url, runtime.get("source", "")) else: _prov = (resolved_provider or self.requested_provider or "").strip() if _prov and _prov != "auto": @@ -173,7 +224,6 @@ class CLIAgentSetupMixin: print("\n⚠️ Provider resolver returned an empty base URL. " "Check your provider config or run: hermes setup") return False - credentials_changed = api_key != self.api_key or base_url != self.base_url routing_changed = resolved_routing != (self.provider, self.api_mode, self.acp_command, self.acp_args) self.provider, self.api_mode, self.acp_command, self.acp_args = resolved_routing @@ -187,8 +237,7 @@ class CLIAgentSetupMixin: # would otherwise send the provider name as the model string -> 400). runtime_model = runtime.get("model") if runtime_model and isinstance(runtime_model, str) and ( - not self.model or self.model == self.provider or self.model == runtime.get("name") - ): + not self.model or self.model == self.provider or self.model == runtime.get("name")): self.model = runtime_model # Still empty (e.g. `hermes auth add` without `hermes model`): fall back to the @@ -201,8 +250,7 @@ class CLIAgentSetupMixin: self.model = _default logger.info( "No model configured — defaulting to %s for provider %s", - _default, resolved_provider, - ) + _default, resolved_provider) except Exception: pass @@ -213,14 +261,17 @@ class CLIAgentSetupMixin: if (credentials_changed or routing_changed or model_changed) and self.agent is not None: self.agent = None self._active_agent_route_signature = None - return True def _resolve_fallback_runtime(self, primary_exc): - """Primary provider auth failed: try each fallback entry in order and switch the - CLI's requested_provider/model to the first that resolves. None if none do.""" + """Primary provider resolution failed: on an AuthError try each fallback entry in + order and switch the CLI's requested_provider/model to the first that resolves. + None when the error is not auth-related or no fallback resolves.""" from cli import _cprint, logger + from hermes_cli.auth import AuthError from hermes_cli.runtime_provider import resolve_runtime_provider + if not isinstance(primary_exc, AuthError): + return None _fb_chain = self._fallback_model if isinstance(self._fallback_model, list) else [] for _fb in _fb_chain: _fb_provider = (_fb.get("provider") or "").strip().lower() @@ -229,7 +280,6 @@ class CLIAgentSetupMixin: continue try: from hermes_cli.fallback_config import resolve_entry_api_key - _fb_kwargs = {"requested": _fb_provider} if _fb.get("base_url"): _fb_kwargs["explicit_base_url"] = _fb["base_url"] @@ -239,8 +289,7 @@ class CLIAgentSetupMixin: runtime = resolve_runtime_provider(**_fb_kwargs) logger.warning( "Primary provider auth failed (%s). Falling through to fallback: %s/%s", - primary_exc, _fb_provider, _fb_model, - ) + primary_exc, _fb_provider, _fb_model) _cprint(f"⚠️ Primary auth failed — switching to fallback: {_fb_provider} / {_fb_model}") self.requested_provider = _fb_provider self.model = _fb_model @@ -255,12 +304,10 @@ class CLIAgentSetupMixin: Never prints or mutates CLI state, so the interactive first-run path can route a keyless install into onboarding before the user types into a chat that can't work.""" from hermes_cli.runtime_provider import resolve_runtime_provider - try: runtime = resolve_runtime_provider( requested=self.requested_provider, explicit_api_key=self._explicit_api_key, - explicit_base_url=self._explicit_base_url, - ) + explicit_base_url=self._explicit_base_url) except Exception: return False if not isinstance(runtime, dict): @@ -276,7 +323,6 @@ class CLIAgentSetupMixin: startup, TTY). Runs the same flow as ``hermes model`` so onboarding has a single source of truth. True when a provider was configured.""" from cli import _cprint, logger - _cprint("") _cprint("⚕ No inference provider is configured yet — let's fix that.") _cprint(" You'll pick a provider (Nous Portal OAuth is the fastest; " @@ -289,7 +335,6 @@ class CLIAgentSetupMixin: if answer in {"n", "no"}: _cprint(" Skipped. Run 'hermes model' or 'hermes setup' any time.") return False - try: from hermes_cli.main import select_provider_and_model select_provider_and_model() @@ -316,7 +361,6 @@ class CLIAgentSetupMixin: # Force credential re-resolution + agent rebuild on next use. self.agent = None self._active_agent_route_signature = None - if self._runtime_credentials_ready(): _cprint(" ✓ Provider configured — you're ready to chat.") return True @@ -328,17 +372,15 @@ class CLIAgentSetupMixin: provider. With `/fast` on (service_tier == "priority") attach request_overrides; auto/cold tiers are applied per request by agent.fast_mode instead.""" from hermes_cli.models import resolve_fast_mode_overrides - runtime = _current_runtime(self) route = {"model": self.model, "runtime": runtime, "signature": _route_signature(self.model, runtime)} overrides = None if getattr(self, "service_tier", None) == "priority": try: overrides = resolve_fast_mode_overrides( - route["model"], provider=runtime["provider"], base_url=runtime["base_url"], - ) + route["model"], provider=runtime["provider"], base_url=runtime["base_url"]) except Exception: - overrides = None + pass route["request_overrides"] = overrides return route @@ -374,7 +416,6 @@ class CLIAgentSetupMixin: print(plain, file=sys.stderr) else: ChatConsole().print(rich) - if not session_meta: if _quiet_mode: print(f"Session not found: {self.session_id}", file=sys.stderr) @@ -388,9 +429,7 @@ class CLIAgentSetupMixin: lambda rid: ChatConsole().print( f"[dim]Session {_escape(self.session_id)} was compressed into " f"{_escape(rid)}; resuming the descendant with your " - f"transcript.[/dim]" - ), - ) + f"transcript.[/dim]")) if getattr(self, "_resume_history_error", None): return False # Only the TIP session's rows are loaded here (no ancestors), so use the @@ -400,8 +439,7 @@ class CLIAgentSetupMixin: self._resume_history_error = resume_limit_error _say( f"Cannot resume session: {resume_limit_error}", - f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}", - ) + f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}") return False restored = self._session_db.get_messages_as_conversation(self.session_id, repair_alternation=True) if restored: @@ -413,8 +451,7 @@ class CLIAgentSetupMixin: _say( f"↻ Resumed session {self.session_id}{title_part} {counts}", f"[bold {_accent_hex()}]↻ Resumed session[/] [bold]{_escape(self.session_id)}[/]" - f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}", - ) + f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}") self._restore_session_cwd(session_meta, quiet=_quiet_mode) self._restore_session_yolo(session_meta, quiet=_quiet_mode) self._restore_session_model(session_meta, quiet=_quiet_mode) @@ -436,20 +473,14 @@ class CLIAgentSetupMixin: # Join the background preloaded-skills load (--skills/-s) BEFORE the agent # snapshots self.system_prompt below. No-op when nothing was requested. self.finalize_preloaded_skills() - _prepare_deferred_agent_startup() self._install_tool_callbacks() self._ensure_tirith_security() - if not self._ensure_runtime_credentials(): return False - from hermes_cli.mcp_startup import ensure_mcp_discovery_before_agent_build - ensure_mcp_discovery_before_agent_build( - logger=logger, single_query=getattr(self, "_single_query_mode", False) - ) - + logger=logger, single_query=getattr(self, "_single_query_mode", False)) if self._session_db is None: try: from hermes_state import SessionDB @@ -458,10 +489,8 @@ class CLIAgentSetupMixin: logger.warning("SQLite session store not available — session will NOT be indexed: %s", e) if ( self._resumed and self._session_db and not self.conversation_history - and not self._load_resumed_history_late() - ): + and not self._load_resumed_history_late()): return False - try: runtime = runtime_override or _current_runtime(self) effective_model = model_override or self.model @@ -470,8 +499,7 @@ class CLIAgentSetupMixin: clarify_callback = ( _single_query_clarify_callback if getattr(self, "_single_query_mode", False) - else self._clarify_callback - ) + else self._clarify_callback) self.agent = AIAgent( model=effective_model, api_key=runtime.get("api_key"), base_url=runtime.get("base_url"), provider=runtime.get("provider"), @@ -507,8 +535,7 @@ class CLIAgentSetupMixin: stream_delta_callback=self._stream_delta if self.streaming_enabled else None, tool_gen_callback=self._on_tool_gen_start if self.streaming_enabled else None, notice_callback=self._on_notice, notice_clear_callback=self._on_notice_clear, - reaction_callback=self._on_reaction, - ) + reaction_callback=self._on_reaction) # Reference for atexit memory-provider shutdown: ``_run_cleanup`` in cli.py # reads ``cli._active_agent_ref``, so this MUST write the ``cli`` module's # global — a ``global`` statement here would bind this module's namespace. @@ -521,7 +548,6 @@ class CLIAgentSetupMixin: # warning shows before the first message. Idempotent + fail-open in the helper. try: from agent.credits_tracker import seed_credits_at_session_start - seed_credits_at_session_start(self.agent) except Exception: pass @@ -544,7 +570,6 @@ class CLIAgentSetupMixin: console = ChatConsole() console.print(f"[bold red]Failed to initialize agent: {e}[/]") from hermes_constants import partial_update_hint - for line in partial_update_hint(e): console.print(line) return False @@ -559,7 +584,6 @@ class CLIAgentSetupMixin: return None from cli import logger from hermes_state import SessionResumeTooLargeError - try: safety_check = getattr(self._session_db, "assert_resume_safe", None) if not callable(safety_check): @@ -573,8 +597,7 @@ class CLIAgentSetupMixin: except Exception as exc: logger.warning( "Resume safety check failed for %s (proceeding without guard): %s", - self.session_id, exc, - ) + self.session_id, exc) return None def _preload_resumed_session(self) -> bool: @@ -584,35 +607,28 @@ class CLIAgentSetupMixin: from cli import _accent_hex if not self._resumed or not self._session_db: return False - session_meta = self._session_db.get_session(self.session_id) if not session_meta: self._console_print(f"[bold red]Session not found: {self.session_id}[/]") self._console_print("[dim]Use a session ID from a previous CLI run (hermes sessions list).[/]") return False - session_meta = self._follow_compression_chain( session_meta, lambda rid: self._console_print( f"[dim]Session {self.session_id} was compressed into " - f"{rid}; resuming the descendant with your transcript.[/]" - ), - ) + f"{rid}; resuming the descendant with your transcript.[/]")) resume_limit_error = self._resume_history_limit_error() if resume_limit_error: self._resume_history_error = resume_limit_error self._console_print(f"[bold red]Cannot resume session:[/] {resume_limit_error}") return False - restored, display_history = self._session_db.get_resume_conversations(self.session_id) accent_color = _accent_hex() if not restored: self._console_print( f"[{accent_color}]Session {self.session_id} found but has no " - f"messages. Starting fresh.[/]" - ) + f"messages. Starting fresh.[/]") return False - restored = [m for m in restored if m.get("role") != "session_meta"] self.conversation_history = restored self._resume_display_history = [m for m in display_history if m.get("role") != "session_meta"] @@ -626,12 +642,10 @@ class CLIAgentSetupMixin: f"[{accent_color}]↻ Resumed session [bold]{self.session_id}[/bold]" f"{title_part} " f"({msg_count} user message{'s' if msg_count != 1 else ''}, " - f"{len(restored)} total messages)[/]" - ) + f"{len(restored)} total messages)[/]") self._restore_session_cwd(session_meta) self._restore_session_yolo(session_meta) self._restore_session_model(session_meta) - self._reopen_session() return True @@ -643,88 +657,22 @@ class CLIAgentSetupMixin: display_history = getattr(self, "_resume_display_history", self.conversation_history) if not display_history or self.resume_display == "minimal": return - _disp = CLI_CONFIG.get("display", {}) - MAX_DISPLAY_EXCHANGES = int(_disp.get("resume_exchanges", 10)) - MAX_USER_LEN = int(_disp.get("resume_max_user_chars", 300)) - MAX_ASST_LEN = int(_disp.get("resume_max_assistant_chars", 200)) - MAX_ASST_LINES = int(_disp.get("resume_max_assistant_lines", 3)) - SKIP_TOOL_ONLY = _disp.get("resume_skip_tool_only", True) - - # Collect displayable (role, text) entries; system and tool-result rows are skipped. - entries = [] - _last_asst_idx = None # index of last assistant entry - _last_asst_full = None # un-truncated display text for last assistant - for msg in display_history: - role = msg.get("role", "") - display_kind = msg.get("display_kind") - content = msg.get("content") - tool_calls = msg.get("tool_calls") or [] - if display_kind == "hidden": - continue - if display_kind in _RESUME_EVENT_TEXT: - entries.append(("event", _RESUME_EVENT_TEXT[display_kind])) - continue - if role in ("system", "tool"): - continue - - # Stored history is untrusted for display: strip escape sequences/control - # chars so replay can't clear the screen, retitle the window or restyle the panel. - if role == "user": - text = _sanitize_display_text(_user_display_text(content)) - if len(text) > MAX_USER_LEN: - text = text[:MAX_USER_LEN] + "..." - entries.append(("user", text)) - - elif role == "assistant": - text = "" if content is None else str(content) - text = _sanitize_display_text(_strip_reasoning_tags(text)) - parts = [] - full_parts = [] # un-truncated version - if text: - full_parts.append(text) - lines = text.splitlines() - if len(lines) > MAX_ASST_LINES: - text = "\n".join(lines[:MAX_ASST_LINES]) + " ..." - if len(text) > MAX_ASST_LEN: - text = text[:MAX_ASST_LEN] + "..." - parts.append(text) - if tool_calls: - parts.append(_tool_calls_summary(tool_calls)) - full_parts.append(parts[-1]) - # Skip pure-reasoning messages with no visible output, and tool-call-only - # entries when SKIP_TOOL_ONLY is enabled. - if not text and (SKIP_TOOL_ONLY or not tool_calls): - continue - entries.append(("assistant", " ".join(parts))) - _last_asst_idx = len(entries) - 1 - _last_asst_full = " ".join(full_parts) - + entries, _last_asst_idx, _last_asst_full = _collect_resume_entries( + display_history, _disp, lambda t: _sanitize_display_text(_strip_reasoning_tags(t))) if not entries: return - - skipped = max(0, len(entries) - MAX_DISPLAY_EXCHANGES * 2) + skipped = max(0, len(entries) - int(_disp.get("resume_exchanges", 10)) * 2) entries = entries[skipped:] - # Show the last assistant entry in full so the user sees where they left off. if _last_asst_idx is not None and _last_asst_full: adj_idx = _last_asst_idx - skipped if 0 <= adj_idx < len(entries): entries[adj_idx] = ("assistant_last", _last_asst_full) - from rich.panel import Panel from rich.text import Text - - try: - from hermes_cli.skin_engine import get_active_skin - _skin = get_active_skin() - _history_text_c, _session_label_c, _session_border_c, _assistant_label_c = ( - _skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS - ) - except Exception: - _history_text_c, _session_label_c, _session_border_c, _assistant_label_c = ( - default for _, default in _RESUME_SKIN_COLORS - ) + _history_text_c, _session_label_c, _session_border_c, _assistant_label_c = ( + _resume_panel_colors()) # role -> (label, label style, body style, continuation indent) role_styles = { @@ -732,11 +680,9 @@ class CLIAgentSetupMixin: "assistant": (" ◆ Hermes: ", f"dim bold {_assistant_label_c}", "dim", " " * 12), "assistant_last": (" ◆ Hermes: ", f"bold {_assistant_label_c}", "", " " * 12), # full, non-dim } - lines = Text() if skipped: lines.append(f" ... {skipped} earlier messages ...\n\n", style="dim italic") - for i, (role, text) in enumerate(entries): if role == "event": lines.append(f" ◈ {text}\n", style="dim italic") @@ -749,11 +695,9 @@ class CLIAgentSetupMixin: lines.append(f"{indent}{ml}\n", style=body_style) if i < len(entries) - 1: lines.append("") # small gap - panel = Panel( lines, title=f"[dim {_session_label_c}]Previous Conversation[/]", - border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c, - ) + border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c) _record_output_history_entry(lambda: self._render_resume_history_panel_lines(panel)) with _suspend_output_history(): self._console_print(panel) diff --git a/hermes_cli/cli_loops_mixin.py b/hermes_cli/cli_loops_mixin.py index c3f91761cb..10aec28cc5 100644 --- a/hermes_cli/cli_loops_mixin.py +++ b/hermes_cli/cli_loops_mixin.py @@ -49,8 +49,7 @@ class CLILoopsMixin: def _cmd_clear(self, cmd_original: str): from cli import ( ChatConsole, _build_compact_banner, _clear_output_history, _cprint, - build_welcome_banner, get_tool_definitions, - ) + build_welcome_banner, get_tool_definitions) if self._confirm_destructive_slash( "clear", "This clears the screen and starts a new session.\n" @@ -78,14 +77,14 @@ class CLILoopsMixin: cc.print(_build_compact_banner()) else: tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, quiet_mode=True) + agent = getattr(self, "agent", None) ctx_len = None - if hasattr(self, 'agent') and self.agent and hasattr(self.agent, 'context_compressor'): - ctx_len = self.agent.context_compressor.context_length + if agent and hasattr(agent, "context_compressor"): + ctx_len = agent.context_compressor.context_length build_welcome_banner( console=cc, model=self.model, cwd=os.getenv("TERMINAL_CWD", os.getcwd()), tools=tools, enabled_toolsets=self.enabled_toolsets, session_id=self.session_id, - context_length=ctx_len, provider=self.provider, - ) + context_length=ctx_len, provider=self.provider) _cprint(_FRESH_START) self._print_random_tip() @@ -179,8 +178,7 @@ class CLILoopsMixin: _undo_desc = ( "This removes the last user/assistant exchange from history." if _undo_n == 1 - else f"This removes the last {_undo_n} user turns from history." - ) + else f"This removes the last {_undo_n} user turns from history.") if self._confirm_destructive_slash("undo", _undo_desc, cmd_original=cmd_original) is None: return True # confirmation cancelled — command handled, keep REPL alive self.undo_last(_undo_n) @@ -191,11 +189,9 @@ class CLILoopsMixin: def _cmd_egress(self, cmd_original: str): from hermes_cli.slash_exec import CommandContext, execute_command - self._console_print( execute_command("egress", CommandContext(surface="cli")).text, - highlight=False, markup=False, - ) + highlight=False, markup=False) def _cmd_statusbar(self, cmd_original: str): self._status_bar_visible = not self._status_bar_visible @@ -208,7 +204,6 @@ class CLILoopsMixin: def _cmd_version(self, cmd_original: str): from hermes_cli.main import _print_version_info - _print_version_info(check_updates=True) def _cmd_reload(self, cmd_original: str): @@ -227,9 +222,7 @@ class CLILoopsMixin: # installed-but-not-enabled plugins show up; the plugin manager only knows # *loaded* plugins and made fresh installs look like "nothing installed". from hermes_cli.plugins_cmd import ( - _discover_all_plugins, _get_disabled_set, _get_enabled_set, _plugin_status - ) - + _discover_all_plugins, _get_disabled_set, _get_enabled_set, _plugin_status) entries = _discover_all_plugins() enabled = _get_enabled_set() disabled = _get_disabled_set() @@ -238,7 +231,6 @@ class CLILoopsMixin: # on one line (full catalog behind `hermes plugins list`). user_entries = [e for e in entries if e[3] != "bundled"] bundled_count = len(entries) - len(user_entries) - if not user_entries: print("No user plugins installed.") print(" Install one: hermes plugins install owner/repo") @@ -246,15 +238,11 @@ class CLILoopsMixin: if bundled_count: print(f" ({bundled_count} bundled plugins available — see: hermes plugins list)") return - # Loaded-plugin details (tools/hooks/commands counts, errors) by name. - loaded: dict = {} - try: + try: # loaded-plugin details (tools/hooks/commands counts, errors) by name from hermes_cli.plugins import get_plugin_manager - for p in get_plugin_manager().list_plugins(): - loaded[p["name"]] = p + loaded = {p["name"]: p for p in get_plugin_manager().list_plugins()} except Exception: loaded = {} - print(f"User plugins ({len(user_entries)}):") for name, version, _desc, source, _dir, key in sorted(user_entries): state = _plugin_status(name, enabled, disabled, key=key) @@ -362,18 +350,13 @@ class CLILoopsMixin: return self._session_bound_manager("_loop_manager", "loop manager", load) def _start_heartbeat_watchdog(self): - """Start the idle-poll thread that fires due heartbeats. - - Same pattern as the wake-word watchdog: a daemon thread polls a few times a - minute; when the session is idle (no agent running, empty input queue) and the - heartbeat is due, its prompt is injected into ``_pending_input`` as a normal user - turn. Missed ticks coalesce — the anchor resets on fire, so a busy hour yields ONE - heartbeat turn. Idempotent; safe to call on every /heartbeat set. - """ + """Start the idle-poll daemon that injects a due heartbeat prompt into + ``_pending_input`` as a normal user turn when the session is idle. Missed ticks + coalesce (the anchor resets on fire, so a busy hour yields ONE heartbeat turn). + Idempotent; safe to call on every /heartbeat set.""" if getattr(self, "_heartbeat_watchdog_started", False): return self._heartbeat_watchdog_started = True - from hermes_cli.heartbeat import POLL_SECONDS def _loop(): @@ -388,8 +371,7 @@ class CLILoopsMixin: self._agent_running or getattr(self, "_voice_recording", False) or getattr(self, "_voice_processing", False) - or not self._pending_input.empty() - ) + or not self._pending_input.empty()) if busy: continue prompt = mgr.due_prompt() @@ -399,7 +381,6 @@ class CLILoopsMixin: logging.debug("heartbeat watchdog tick failed: %s", exc) finally: self._heartbeat_watchdog_started = False - threading.Thread(target=_loop, daemon=True, name="heartbeat-watchdog").start() def _maybe_fire_loop_tick(self) -> None: @@ -426,12 +407,10 @@ class CLILoopsMixin: return try: from hermes_cli.loops import goal_blocks_loop_tick - if goal_blocks_loop_tick(mgr.session_id): return except Exception: pass - wakeup = mgr.fire_tick() if not wakeup: return @@ -470,8 +449,7 @@ class CLILoopsMixin: parts = [ p.get("text", "") for p in content - if isinstance(p, dict) and p.get("type") in {"text", "output_text"} - ] + if isinstance(p, dict) and p.get("type") in {"text", "output_text"}] return "\n".join(t for t in parts if t) return str(content or "") except Exception: @@ -502,10 +480,8 @@ class CLILoopsMixin: pass _cprint( f" {_DIM}⏸ Loop paused — wakeup turn was interrupted. " - f"Use /loop resume to continue, or /loop stop to end it.{_RST}" - ) + f"Use /loop resume to continue, or /loop stop to end it.{_RST}") return - decision = mgr.complete_tick(self._last_assistant_response_text()) msg = decision.get("message") or "" if msg: @@ -514,15 +490,11 @@ class CLILoopsMixin: _cprint(f" {_DIM}↻ Loop: {mgr.state.remaining_label()}.{_RST}") def _maybe_continue_goal_after_turn(self) -> None: - """Hook run after every CLI turn: judge the goal and maybe re-queue a continuation. - - Safe when no goal is set. Preemption is automatic: a real user message already in - ``_pending_input`` skips judging (the user's turn goes first; we re-judge after). - A user-cancelled turn (Ctrl+C) AUTO-PAUSES the goal instead of judging — the judge - would run on partial output, almost always say "continue", and re-queue exactly - what the user cancelled; pausing is observable and recoverable via ``/goal - resume``. The empty-response skip mirrors ``_handle_message`` in ``gateway/run.py``. - """ + """Post-turn hook: judge the goal and maybe re-queue a continuation. A real user + message already queued preempts judging (re-judged after their turn). Ctrl+C + AUTO-PAUSES instead of judging — the judge on partial output nearly always says + "continue" and would re-queue exactly what was cancelled; pausing is recoverable + via ``/goal resume``. Empty-response skip mirrors ``gateway/run.py``.""" from cli import _DIM, _RST, _cprint, _looks_like_slash_command mgr = self._get_goal_manager() if mgr is None or not mgr.is_active(): @@ -551,7 +523,6 @@ class CLILoopsMixin: return except Exception: pass - if getattr(self, "_last_turn_interrupted", False): try: mgr.pause(reason="user-interrupted (Ctrl+C)") @@ -559,8 +530,7 @@ class CLILoopsMixin: logging.debug("goal pause-on-interrupt failed: %s", exc) _cprint( f" {_DIM}⏸ Goal paused — turn was interrupted. " - f"Use /goal resume to continue, or /goal clear to stop.{_RST}" - ) + f"Use /goal resume to continue, or /goal clear to stop.{_RST}") return # Empty/whitespace responses are almost always transient failures (API error, @@ -568,20 +538,16 @@ class CLILoopsMixin: last_response = self._last_assistant_response_text() if not last_response.strip(): return - try: from hermes_cli.goals import gather_background_processes as _gather_bg _bg_procs = _gather_bg() except Exception: _bg_procs = None - decision = mgr.evaluate_after_turn( - last_response, user_initiated=True, background_processes=_bg_procs - ) + last_response, user_initiated=True, background_processes=_bg_procs) msg = decision.get("message") or "" if msg: _cprint(f" {msg}") - if decision.get("should_continue"): prompt = decision.get("continuation_prompt") if prompt: diff --git a/hermes_cli/cli_terminal_mixin.py b/hermes_cli/cli_terminal_mixin.py index ed522b418d..5209de1d0a 100644 --- a/hermes_cli/cli_terminal_mixin.py +++ b/hermes_cli/cli_terminal_mixin.py @@ -65,8 +65,7 @@ class CLITerminalMixin: pass logger.warning( "Terminal I/O broken%s — freezing UI paints to avoid redraw storm (#81521)", - f" ({reason})" if reason else "", - ) + f" ({reason})" if reason else "") def _app_invalidate(self, app, where: str, *, swallow: bool) -> None: """``app.invalidate()``: EIO freezes paints, other OSErrors re-raise, and any @@ -83,17 +82,13 @@ class CLITerminalMixin: raise def _invalidate(self, min_interval: float = 0.25) -> None: - """Throttled UI repaint for high-frequency background updates. + """Throttled UI repaint for high-frequency background updates (spinner frames, + streaming flushes): the throttle prevents blinking on slow/SSH links and the + resize-recovery guard keeps footer chrome out of scrollback mid-SIGWINCH. - For spinner frames, streaming token flushes and other repaints that fire many - times per second: the throttle prevents blinking on slow/SSH links, and the - resize-recovery guard avoids stamping footer/status-bar chrome into scrollback - while a SIGWINCH reflow is in flight. - - Do NOT use for user-blocking modal prompts (approval / clarify / sudo): those - must paint immediately via ``_paint_now``. Sent through this throttle, an - unrelated repaint within the 250ms window — or an in-flight resize — silently - drops the modal's entry paint, so it never renders and times out unseen (#41098). + NOT for user-blocking modals (approval / clarify / sudo) — use ``_paint_now``: a + throttled or resize-gated entry paint is silently dropped, so the prompt never + renders and times out unseen (#41098). """ if getattr(self, "_terminal_io_broken", False): return @@ -139,14 +134,10 @@ class CLITerminalMixin: self._app_invalidate(app, "force_full_redraw", swallow=True) def _schedule_focus_regain_redraw(self, min_interval: float = 1.0) -> None: - """Repaint after a terminal focus-in report (``CSI I``), rate-limited. - - Terminals with focus tracking (Ghostty, iTerm2, xterm, muxes toggling DECSET - 1004) emit ``\\x1b[I`` when the Hermes tab becomes visible again; emulators may - coalesce hidden-tab output, so on regain the incremental diff stacks on stale - content (#60920 focus-regain variant, #25337). Self-gating — terminals without - focus tracking never emit it. Rate-limited so a burst (rapid Alt+Tab, pane hops) - repaints at most once per ``min_interval`` seconds. + """Repaint after a terminal focus-in report (``CSI I``), at most once per + ``min_interval`` (rapid Alt+Tab / pane-hop bursts). Emulators with focus tracking + (DECSET 1004) may coalesce hidden-tab output, so on regain the incremental diff + stacks on stale content (#60920, #25337); terminals without it never emit ``CSI I``. """ now = time.monotonic() if now - getattr(self, "_last_focus_regain_redraw", 0.0) < min_interval: @@ -172,18 +163,12 @@ class CLITerminalMixin: return bool(raw) def _recover_terminal_after_interrupt(self) -> None: - """Recover the terminal after an interrupted agent turn (#33271). - - An in-flight ``CSI 6n`` cursor query may answer (``ESC[;R``) after the - input parser tore down: the reply leaks as literal text and the VT100 parser can - stall mid-escape, so the terminal looks frozen. ``flush_stdin()`` drains stray - bytes (no-op on non-TTY), then ``_force_full_redraw()`` repaints cleanly; each - step self-guards so one failing never blocks the other. A dead PTY (EIO) skips - the redraw — painting a broken fd is the #81521 redraw storm. - - Do NOT clear output history here: the interruption marker is printed under - ``_suspend_output_history`` in chat(), so the replay reproduces the response - without duplicating the marker (#60920). + """Recover the terminal after an interrupted agent turn (#33271): an in-flight + ``CSI 6n`` reply arriving after the input parser tore down leaks as literal text + and can stall the VT100 parser mid-escape. Drain stdin, then full redraw; each step + self-guards. A dead PTY (EIO) skips the redraw (#81521 redraw storm). Never clear + output history here — the interruption marker is printed under + ``_suspend_output_history``, so replay already excludes it (#60920). """ if getattr(self, "_terminal_io_broken", False): return @@ -222,30 +207,20 @@ class CLITerminalMixin: def _recover_after_resize(self, app, original_on_resize) -> None: """Recover a resized classic CLI without desynchronizing cursor state. - Unlike ``_force_full_redraw`` this never clears scrollback: the startup banner - lives there and ``_replay_output_history`` cannot reconstruct it. prompt_toolkit's - own resize path runs with its renderer cursor cache intact — its - ``Application._on_resize()`` erases via the cached cursor position, and resetting - the renderer first loses the origin and strands stale prompt glyphs. - - ``_status_bar_suppressed_after_resize`` hides the dynamic status bar / input rules - while the reflow settles: on column shrink the terminal reflows already-painted - rows into scrollback before prompt_toolkit erases them, so a fresh bar looks - duplicated (#19280, #22976). Suppression alone cannot erase an already-reflowed - OLD bar (``renderer.erase()`` does ``cursor_up(_cursor_pos.y)`` with the y cached - at the OLD width, undershooting the reflowed rows), so on an OBSERVED width - change we wipe the viewport (CSI 2J — banner-safe; 3J only when - ``display.cli_rebuild_scrollback_on_redraw``) and replay the transcript before - delegating. Same-width SIGWINCH (tmux attach, GNOME tab bar, focus signals) and - the first signal without a baseline are left untouched: a 2J+replay against - preserved scrollback duplicates everything in ``_OUTPUT_HISTORY`` (#65293); - ``_install_resize_recovery`` seeds the baseline so an initial maximize still - counts. The stale-previous_screen crash on tmux attach is handled by - ``_hermes_call_output_screen_diff``'s retry (#83874). - - Suppression is transient: a debounced timer clears it and repaints once the - reflow settles, so the bar returns during idle (previously it stayed hidden - until the next submitted input). The next-submit clear remains as a fast path. + Never clears scrollback (the startup banner lives there and replay cannot rebuild + it) and never resets the renderer before prompt_toolkit's own ``_on_resize``, + which erases via the cached cursor position. The status bar / input rules are + suppressed while the reflow settles: on column shrink the terminal reflows + already-painted rows into scrollback first, so a fresh bar looks duplicated + (#19280, #22976). Suppression cannot erase the already-reflowed OLD bar + (``renderer.erase()`` uses ``_cursor_pos.y`` cached at the OLD width), so on an + OBSERVED width change we wipe the viewport (CSI 2J, banner-safe; 3J only via + ``display.cli_rebuild_scrollback_on_redraw``) and replay the transcript first. + Same-width SIGWINCH (tmux attach, GNOME tab bar, focus) and the first signal + without a seeded baseline are left alone — 2J+replay against preserved scrollback + duplicates ``_OUTPUT_HISTORY`` (#65293). tmux-attach's stale previous_screen is + handled by ``_hermes_call_output_screen_diff`` (#83874). Suppression is cleared by + a debounced timer so the bar returns during idle; next-submit stays a fast path. """ from cli import _replay_output_history self._status_bar_suppressed_after_resize = True @@ -258,8 +233,7 @@ class CLITerminalMixin: if width_changed: try: self._clear_prompt_toolkit_screen( - app, rebuild_scrollback=self._redraw_rebuilds_scrollback() - ) + app, rebuild_scrollback=self._redraw_rebuilds_scrollback()) _replay_output_history() except Exception: pass @@ -298,10 +272,8 @@ class CLITerminalMixin: app.invalidate() except Exception: pass - self._restart_debounce_timer( - "_status_bar_unsuppress_timer", delay, lambda _t: _run_on_app_loop(app, _clear) - ) + "_status_bar_unsuppress_timer", delay, lambda _t: _run_on_app_loop(app, _clear)) except Exception: # Fail open: never leave the bar stuck hidden. self._status_bar_suppressed_after_resize = False @@ -322,9 +294,7 @@ class CLITerminalMixin: self._resize_recovery_timer = None self._resize_recovery_pending = False self._recover_after_resize(app, original_on_resize) - _run_on_app_loop(app, _run_recovery) - with lock: self._resize_recovery_pending = True self._restart_debounce_timer("_resize_recovery_timer", delay, _timer_fired) @@ -334,16 +304,12 @@ class CLITerminalMixin: def _install_resize_recovery(self, app) -> None: """Route ``app._on_resize`` through the debounced ghost-clearing recovery - (#5474/#49120) and seed the width baseline for width-change detection. - - The seed keeps the session's FIRST SIGWINCH honest (#65293): without it a benign - signal is indistinguishable from a real resize. It reads ``app.output`` directly, - NOT ``_get_tui_terminal_width``: before ``app.run()`` ``get_app()`` is the - DummyApplication whose DummyOutput reports a hardcoded 80 columns, and seeding - that fake width would make the first real signal look like a change. - ``app.output`` is what the running resize handler measures, so both are comparable. + (#5474/#49120) and seed the width baseline so the FIRST SIGWINCH can tell a benign + signal from a real resize (#65293). Reads ``app.output`` directly, NOT + ``_get_tui_terminal_width``: before ``app.run()`` the DummyApplication's output + reports a hardcoded 80 columns, which would make the first real signal look like a + width change. ``app.output`` is what the running resize handler measures. """ - width = None try: width = app.output.get_size().columns except Exception: @@ -358,14 +324,12 @@ class CLITerminalMixin: def _resize_clear_ghosts(): self._schedule_resize_recovery(app, original_on_resize) - app._on_resize = _resize_clear_ghosts def _try_attach_clipboard_image(self) -> bool: """Save a clipboard image to ~/.hermes/images/ and attach it; True if attached.""" from cli import datetime from hermes_cli.clipboard import save_clipboard_image - self._image_counter += 1 ts = datetime.now().strftime("%Y%m%d_%H%M%S") img_path = get_hermes_home() / "images" / f"clip_{ts}_{self._image_counter}.png" @@ -393,14 +357,12 @@ class CLITerminalMixin: """Best-effort reset when leaked mouse reports indicate mode drift.""" from cli import ( CLI_CONFIG, _DIM, _RST, _TERMINAL_INPUT_MODE_RESET_SEQ, - _cli_multiline_shortcuts_enabled, _cprint, _enable_extended_enter_keys, logger, - ) + _cli_multiline_shortcuts_enabled, _cprint, _enable_extended_enter_keys, logger) now = time.monotonic() # Rate-limit to avoid thrashing if a terminal floods reports. if now - self._last_input_mode_recovery < 0.5: return self._last_input_mode_recovery = now - app = getattr(self, "_app", None) output = getattr(app, "output", None) if app else None try: @@ -415,24 +377,19 @@ class CLITerminalMixin: _enable_extended_enter_keys(output) except Exception: pass - logger.warning("Recovered terminal input modes after leak: %s", reason) if not self._input_mode_recovery_notice_shown: self._input_mode_recovery_notice_shown = True _cprint( f" {_DIM}Recovered terminal input modes after leaked mouse reports. " - f"If this repeats, run /new or restart this tab.{_RST}" - ) + f"If this repeats, run /new or restart this tab.{_RST}") def _check_termios_drift(self) -> None: - """Watchdog: heal the tty if it drifted back to cooked mode. - - See ``_heal_cooked_mode_drift``: a lost ``run_in_terminal`` cooked→raw restore - leaves the tty line-buffering while prompt_toolkit believes it owns raw mode — - the CLI looks dead but the process is healthy. Called from ``process_loop``'s - idle branch so it self-heals within ~1s of idling. Skipped while a - ``run_in_terminal`` window legitimately holds cooked mode, while the agent is - running (approval/sudo prompts manipulate the tty), and on Windows (no termios). + """Idle watchdog: heal a tty that drifted back to cooked mode (a lost + ``run_in_terminal`` cooked→raw restore leaves it line-buffering while prompt_toolkit + believes it owns raw mode — CLI looks dead, process is healthy). Skipped while + ``run_in_terminal`` legitimately holds cooked mode, while the agent runs + (approval/sudo prompts touch the tty), and on Windows (no termios). """ from cli import _DIM, _RST, _cprint, _heal_cooked_mode_drift, logger if os.name == "nt": @@ -455,8 +412,7 @@ class CLITerminalMixin: if _heal_cooked_mode_drift(fd): logger.warning( "Healed cooked-mode termios drift on stdin — a " - "run_in_terminal cooked→raw restore was lost." - ) + "run_in_terminal cooked→raw restore was lost.") try: self._invalidate() # so the prompt is visibly alive again except Exception: @@ -465,5 +421,4 @@ class CLITerminalMixin: self._termios_drift_notice_shown = True _cprint( f" {_DIM}Recovered terminal from cooked-mode drift " - f"(input should respond normally again).{_RST}" - ) + f"(input should respond normally again).{_RST}") diff --git a/hermes_cli/clipboard.py b/hermes_cli/clipboard.py index 3e7b82b384..50b6ed9843 100644 --- a/hermes_cli/clipboard.py +++ b/hermes_cli/clipboard.py @@ -177,8 +177,7 @@ _PS_IMAGE_STRATEGIES = ( "if ($null -eq $img) { exit 1 }" "$ms = New-Object System.IO.MemoryStream;" "$img.Save($ms, [System.Drawing.Imaging.ImageFormat]::Png);" - "[System.Convert]::ToBase64String($ms.ToArray())", - ), + "[System.Convert]::ToBase64String($ms.ToArray())"), ( # Get-Clipboard -Format Image (System.Drawing.Image or WPF BitmapSource) "try { " "$img = Get-Clipboard -Format Image -ErrorAction Stop;" @@ -199,8 +198,7 @@ _PS_IMAGE_STRATEGIES = ( "$enc.Save($ms)" "} else { exit 2 }" "[System.Convert]::ToBase64String($ms.ToArray())" - "} catch { exit 1 }", - ), + "} catch { exit 1 }"), ( # copied image *file* (Explorer file drop) _PS_FILEDROP_HIT + "if ($null -ne $hit) { 'True' } else { 'False' }" @@ -322,7 +320,6 @@ def _convert_to_png(path: Path) -> bool: pass except Exception as e: logger.debug("Pillow BMP→PNG conversion failed: %s", e) - tmp = path.with_suffix(".bmp") try: path.rename(tmp) diff --git a/hermes_cli/codex_runtime_plugin_migration.py b/hermes_cli/codex_runtime_plugin_migration.py index 66878cdcfc..dbef3dcea1 100644 --- a/hermes_cli/codex_runtime_plugin_migration.py +++ b/hermes_cli/codex_runtime_plugin_migration.py @@ -15,8 +15,7 @@ logger = logging.getLogger(__name__) # Marker comments wrapping the managed section so re-runs can detect what's ours and what's # user-edited. Both must appear or strip is a no-op. MIGRATION_MARKER = ( - "# managed by hermes-agent — `hermes codex-runtime migrate` regenerates this section" -) + "# managed by hermes-agent — `hermes codex-runtime migrate` regenerates this section") MIGRATION_END_MARKER = "# end hermes-agent managed section" @@ -65,8 +64,7 @@ _KNOWN_HERMES_KEYS = { "command", "args", "env", "cwd", "url", "headers", "transport", "timeout", "connect_timeout", - "enabled", "description", -} + "enabled", "description"} _KEYS_DROPPED_WITH_WARNING = {"sampling"} # (hermes key, codex key, skip note) — timeouts are emitted as floats or skipped when non-numeric. @@ -91,7 +89,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict], """ if not isinstance(hermes_cfg, dict): return None, [] - skipped: list[str] = [] out: dict[str, Any] = {} if hermes_cfg.get("command"): @@ -112,7 +109,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict], skipped.append("transport=sse (codex auto-negotiates)") else: return None, ["no command or url field"] - for hermes_key, codex_key, note in _TIMEOUT_KEYS: if hermes_key in hermes_cfg: try: @@ -121,7 +117,6 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict], skipped.append(note) if hermes_cfg.get("enabled") is False: out["enabled"] = False - for key in hermes_cfg: if key in _KEYS_DROPPED_WITH_WARNING: skipped.append(f"{key} (no codex equivalent)") @@ -171,8 +166,7 @@ def _quote_key(key: str) -> str: def render_codex_toml_section( servers: dict[str, dict], plugins: Optional[list[dict]] = None, - default_permission_profile: Optional[str] = None, -) -> str: + default_permission_profile: Optional[str] = None) -> str: """Render the managed [mcp_servers.] / [plugins.] / default_permissions block. ``default_permission_profile`` (e.g. "workspace-write", "read-only", "full-access") is written @@ -185,7 +179,6 @@ def render_codex_toml_section( out.append("# (no MCP servers, plugins, or permissions configured by Hermes)") out.append(MIGRATION_END_MARKER) return "\n".join(out) + "\n" - if default_permission_profile: profile = default_permission_profile normalized = profile if profile.startswith(":") else f":{profile}" @@ -211,13 +204,11 @@ def _insert_managed_block_at_top_level(user_text: str, managed_block: str) -> st """ if not user_text.strip(): return managed_block - lines = user_text.splitlines(keepends=True) first_table_idx = next((i for i, ln in enumerate(lines) if ln.lstrip().startswith("[")), None) if first_table_idx is None: prefix = user_text.rstrip("\n") return f"{prefix}\n\n{managed_block}" if prefix else managed_block - prefix = "".join(lines[:first_table_idx]).rstrip("\n") suffix = "".join(lines[first_table_idx:]).lstrip("\n") return f"{prefix}\n\n{managed_block}\n{suffix}" if prefix else f"{managed_block}\n{suffix}" @@ -287,30 +278,23 @@ def _strip_existing_managed_block(toml_text: str) -> str: def _query_codex_plugins( - codex_home: Optional[Path] = None, timeout: float = 8.0 -) -> tuple[list[dict], Optional[str]]: - """Query codex's `plugin/list` for installed curated plugins. - - Spawns ``codex app-server`` briefly, sends initialize + plugin/list and keeps installed=true. - Returns ``(plugins, error)``; any failure (not installed, RPC error, timeout) yields ``([], - error)`` and is non-fatal — MCP servers and permissions still write through. - - Plugins codex reports as unavailable (broken install, missing OAuth, removed from marketplace) - are skipped: we write codex's config.toml directly, so a broken plugin would surface as a - codex error on the user's first turn. ``enabled`` is carried forward as codex reports it. + codex_home: Optional[Path] = None, timeout: float = 8.0) -> tuple[list[dict], Optional[str]]: + """Spawn ``codex app-server`` briefly and return ``(installed plugins, error)`` from + ``plugin/list``. Any failure yields ``([], error)`` and is non-fatal (servers and + permissions still write). Plugins codex reports unavailable (broken install, missing OAuth, + delisted) are skipped — we write config.toml directly, so they would surface as a codex + error on the first turn. ``enabled`` is carried forward as reported. """ try: from agent.transports.codex_app_server import CodexAppServerClient except Exception as exc: return [], f"transport unavailable: {exc}" - try: with CodexAppServerClient(codex_home=str(codex_home) if codex_home else None) as client: client.initialize(client_name="hermes-migration") resp = client.request("plugin/list", {}, timeout=timeout) except Exception as exc: return [], f"plugin/list query failed: {exc}" - marketplaces = resp.get("marketplaces") or [] if not isinstance(marketplaces, list): return [], "plugin/list response missing 'marketplaces'" @@ -353,20 +337,16 @@ def _looks_like_test_tempdir(path: str) -> bool: def _build_hermes_tools_mcp_entry() -> dict: - """Build the codex stdio entry that launches Hermes' own tool surface as an MCP server, so the - codex subprocess can call back for browser/web/delegate_task/vision/memory/skills tools. + """Codex stdio entry launching Hermes' own tool surface as an MCP server (browser/web/ + delegate_task/vision/memory/skills call-backs). - HERMES_HOME passes through IF SET so the subprocess sees the same config/auth/sessions DB as - the parent CLI. Read from os.environ (not get_hermes_home()) on purpose: when unset, the codex - subprocess should inherit whatever HERMES_HOME its launcher sets at runtime (systemd unit, - gateway, kanban dispatcher, custom shell) rather than a migrate-time default burned into - config.toml, which would pin it to the wrong profile. The pytest-tempdir guard stops a sibling - test's ``monkeypatch.setenv("HERMES_HOME", tmp_path)`` from leaking a transient dir into the - user's real config.toml. PYTHONPATH passes through so a worktree-launched hermes finds the - branch's modules instead of the installed package. + HERMES_HOME passes through only IF SET, read from os.environ (not get_hermes_home()): when + unset the codex subprocess must inherit its launcher's runtime HERMES_HOME (systemd, gateway, + kanban), not a migrate-time default burned into config.toml that pins the wrong profile. The + pytest-tempdir guard keeps a sibling test's monkeypatched HERMES_HOME out of the user's real + config. PYTHONPATH passes through so a worktree-launched hermes finds the branch's modules. """ import sys - env: dict[str, str] = {} hermes_home = os.environ.get("HERMES_HOME") or "" if hermes_home and not _looks_like_test_tempdir(hermes_home): @@ -376,22 +356,19 @@ def _build_hermes_tools_mcp_entry() -> dict: # Quiet mode + redaction defaults so the MCP wire stays clean. env["HERMES_QUIET"] = "1" env["HERMES_REDACT_SECRETS"] = env.get("HERMES_REDACT_SECRETS", "true") - return { "command": sys.executable, "args": ["-m", "agent.transports.hermes_tools_mcp_server"], "env": env, # Generous timeouts — browser_navigate or delegate_task can take a while. "startup_timeout_sec": 30.0, - "tool_timeout_sec": 600.0, - } + "tool_timeout_sec": 600.0} def _write_atomic(target: Path, text: str) -> None: """Write via a same-directory temp file + rename (atomic on POSIX, ReplaceFile on Windows) so a crash mid-write never leaves a half-written config.toml that codex would refuse to load.""" import tempfile - tmp_fd, tmp_path_str = tempfile.mkstemp(prefix=".config.toml.", dir=str(target.parent)) tmp_path = Path(tmp_path_str) try: @@ -400,8 +377,7 @@ def _write_atomic(target: Path, text: str) -> None: tmp_path.replace(target) except Exception: try: - if tmp_path.exists(): - tmp_path.unlink() + tmp_path.unlink(missing_ok=True) except Exception: pass raise @@ -410,8 +386,7 @@ def _write_atomic(target: Path, text: str) -> None: def migrate( hermes_config: dict, *, codex_home: Optional[Path] = None, dry_run: bool = False, discover_plugins: bool = True, default_permission_profile: Optional[str] = ":workspace", - expose_hermes_tools: bool = True, -) -> MigrationReport: + expose_hermes_tools: bool = True) -> MigrationReport: """Translate Hermes mcp_servers config + Codex curated plugins into ~/.codex/config.toml. ``discover_plugins`` spawns the live codex CLI (set False in tests); discovery is best-effort @@ -425,12 +400,10 @@ def migrate( codex_home = codex_home or Path.home() / ".codex" target = codex_home / "config.toml" report.target_path = target - hermes_servers = (hermes_config or {}).get("mcp_servers") or {} if not isinstance(hermes_servers, dict): report.errors.append("mcp_servers in Hermes config is not a dict; cannot migrate.") return report - translated: dict[str, dict] = {} for raw_name, cfg in hermes_servers.items(): name = str(raw_name) @@ -443,7 +416,6 @@ def migrate( if skipped: report.skipped_keys_per_server[name] = skipped report.migrated.append(name) - plugins: list[dict] = [] plugin_query_succeeded = False if discover_plugins and not dry_run: @@ -454,17 +426,14 @@ def migrate( # re-render and may strip pre-existing tables outside the managed block. plugin_query_succeeded = not plugin_err report.migrated_plugins += [f"{p['name']}@{p['marketplace']}" for p in plugins] - if default_permission_profile: report.wrote_permissions_default = default_permission_profile if expose_hermes_tools: translated["hermes-tools"] = _build_hermes_tools_mcp_entry() if "hermes-tools" not in report.migrated: report.migrated.append("hermes-tools") - managed_block = render_codex_toml_section( - translated, plugins=plugins, default_permission_profile=default_permission_profile, - ) + translated, plugins=plugins, default_permission_profile=default_permission_profile) new_text = managed_block if target.exists(): try: @@ -476,7 +445,6 @@ def migrate( if plugin_query_succeeded: without_managed = _strip_unmanaged_plugin_tables(without_managed) new_text = _insert_managed_block_at_top_level(without_managed, managed_block) - if dry_run: return report try: From 48ef059a6c57ca3234f954fe0dd082f4614dd358 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:35:44 -0700 Subject: [PATCH 2/2] refactor(hermes_cli): dedupe decision-message printing, width-probe loop, plugin dedupe via dict --- hermes_cli/cli_agent_setup_mixin.py | 5 ++- hermes_cli/cli_loops_mixin.py | 39 +++++++++----------- hermes_cli/cli_terminal_mixin.py | 17 ++++----- hermes_cli/codex_runtime_plugin_migration.py | 14 +++---- 4 files changed, 34 insertions(+), 41 deletions(-) diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index afda620ee5..d36056e15f 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -417,12 +417,13 @@ class CLIAgentSetupMixin: else: ChatConsole().print(rich) if not session_meta: + hint = "Use a session ID from a previous CLI run (hermes sessions list)." if _quiet_mode: print(f"Session not found: {self.session_id}", file=sys.stderr) - print("Use a session ID from a previous CLI run (hermes sessions list).", file=sys.stderr) + print(hint, file=sys.stderr) else: _cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}") - _cprint(f"{_DIM}Use a session ID from a previous CLI run (hermes sessions list).{_RST}") + _cprint(f"{_DIM}{hint}{_RST}") return False session_meta = self._follow_compression_chain( session_meta, diff --git a/hermes_cli/cli_loops_mixin.py b/hermes_cli/cli_loops_mixin.py index 10aec28cc5..2952479ebf 100644 --- a/hermes_cli/cli_loops_mixin.py +++ b/hermes_cli/cli_loops_mixin.py @@ -21,6 +21,15 @@ def _preview(payload: str) -> str: return f"{payload[:80]}{'...' if len(payload) > 80 else ''}" +def _print_decision_message(decision: dict) -> bool: + """Print a manager decision's ``message`` (if any) via _cprint; True when one was printed.""" + from cli import _cprint + msg = decision.get("message") or "" + if msg: + _cprint(f" {msg}") + return bool(msg) + + class CLILoopsMixin: """Simple slash-command wrappers plus goal/heartbeat/loop manager hooks for the interactive CLI""" @@ -246,10 +255,10 @@ class CLILoopsMixin: print(f"User plugins ({len(user_entries)}):") for name, version, _desc, source, _dir, key in sorted(user_entries): state = _plugin_status(name, enabled, disabled, key=key) - glyph = {"enabled": "✓", "disabled": "✗"}.get(state, "○") - ver = f" v{version}" if version else "" info = loaded.get(name) or {} bits = [f"{info[k]} {k}" for k in ("tools", "hooks", "commands") if info.get(k)] + glyph = {"enabled": "✓", "disabled": "✗"}.get(state, "○") + ver = f" v{version}" if version else "" detail = f" ({', '.join(bits)})" if bits else "" label = "" if state == "enabled" else f" [{state}]" error = f" — {info['error']}" if info.get("error") else "" @@ -432,10 +441,7 @@ class CLILoopsMixin: # judge; complete them immediately (caps and scheduling still apply). if wakeup.lstrip().startswith("/"): try: - decision = mgr.complete_tick("") - msg = decision.get("message") or "" - if msg: - _cprint(f" {msg}") + _print_decision_message(mgr.complete_tick("")) except Exception: pass @@ -483,10 +489,8 @@ class CLILoopsMixin: f"Use /loop resume to continue, or /loop stop to end it.{_RST}") return decision = mgr.complete_tick(self._last_assistant_response_text()) - msg = decision.get("message") or "" - if msg: - _cprint(f" {msg}") - elif decision.get("status") == "active" and mgr.state is not None: + if (not _print_decision_message(decision) and decision.get("status") == "active" + and mgr.state is not None): _cprint(f" {_DIM}↻ Loop: {mgr.state.remaining_label()}.{_RST}") def _maybe_continue_goal_after_turn(self) -> None: @@ -508,15 +512,10 @@ class CLILoopsMixin: try: pending = getattr(self, "_pending_input", None) if pending is not None and not pending.empty(): - has_real_message = False try: - for entry in list(pending.queue): - if isinstance(entry, tuple) and entry: - entry = entry[0] - if isinstance(entry, str) and _looks_like_slash_command(entry): - continue - has_real_message = True - break + entries = [e[0] if isinstance(e, tuple) and e else e for e in list(pending.queue)] + has_real_message = any( + not (isinstance(e, str) and _looks_like_slash_command(e)) for e in entries) except Exception: has_real_message = True # can't introspect — defer to be safe if has_real_message: @@ -545,9 +544,7 @@ class CLILoopsMixin: _bg_procs = None decision = mgr.evaluate_after_turn( last_response, user_initiated=True, background_processes=_bg_procs) - msg = decision.get("message") or "" - if msg: - _cprint(f" {msg}") + _print_decision_message(decision) if decision.get("should_continue"): prompt = decision.get("continuation_prompt") if prompt: diff --git a/hermes_cli/cli_terminal_mixin.py b/hermes_cli/cli_terminal_mixin.py index 5209de1d0a..76bb873831 100644 --- a/hermes_cli/cli_terminal_mixin.py +++ b/hermes_cli/cli_terminal_mixin.py @@ -310,21 +310,18 @@ class CLITerminalMixin: reports a hardcoded 80 columns, which would make the first real signal look like a width change. ``app.output`` is what the running resize handler measures. """ - try: - width = app.output.get_size().columns - except Exception: - width = None - if not width or width <= 0: + width = None + for probe in (lambda: app.output.get_size().columns, + lambda: shutil.get_terminal_size((80, 24)).columns): try: - width = shutil.get_terminal_size((80, 24)).columns + width = probe() except Exception: width = None + if width and width > 0: + break self._last_resize_width = width original_on_resize = app._on_resize - - def _resize_clear_ghosts(): - self._schedule_resize_recovery(app, original_on_resize) - app._on_resize = _resize_clear_ghosts + app._on_resize = lambda: self._schedule_resize_recovery(app, original_on_resize) def _try_attach_clipboard_image(self) -> bool: """Save a clipboard image to ~/.hermes/images/ and attach it; True if attached.""" diff --git a/hermes_cli/codex_runtime_plugin_migration.py b/hermes_cli/codex_runtime_plugin_migration.py index dbef3dcea1..ab49d9d529 100644 --- a/hermes_cli/codex_runtime_plugin_migration.py +++ b/hermes_cli/codex_runtime_plugin_migration.py @@ -298,8 +298,7 @@ def _query_codex_plugins( marketplaces = resp.get("marketplaces") or [] if not isinstance(marketplaces, list): return [], "plugin/list response missing 'marketplaces'" - out: list[dict] = [] - seen: set[tuple[str, str]] = set() + out: dict[tuple[str, str], dict] = {} # (name, marketplace) -> entry; first wins for marketplace in marketplaces: if not isinstance(marketplace, dict): continue @@ -314,12 +313,11 @@ def _query_codex_plugins( availability) continue name = str(plugin.get("name") or "") - if not name or (name, market_name) in seen: - continue - seen.add((name, market_name)) - out.append({"name": name, "marketplace": market_name, - "enabled": bool(plugin.get("enabled", True))}) - return out, None + if name: + out.setdefault((name, market_name), { + "name": name, "marketplace": market_name, + "enabled": bool(plugin.get("enabled", True))}) + return list(out.values()), None # pytest tempdir shapes: ``pytest-of-/pytest-/``, macOS ``/private/var/folders/…/T``.