hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 33 Packages Projects Releases Wiki Activity
Files
e389aa40615193ef03156074b53fb549959cb277
hermes-agent/cli.py
teknium1 e389aa4061 fix(cli): close resize-drag row loss and modal refill dups (#95375) 
- Gate every paint and chrome render on the live width: a paint or render that
  would land after the width changed but before its recovery now waits for it
  (checked when it actually runs on the loop, not when it was scheduled).
- Recovery that sees the width move again keeps holding; the next recovery runs
  right after the next width change once the 0.85s hold cap passed.
- Track the narrowest width of a drag: narrowing and widening back still refills.
- Chrome floor: after a refill the chrome is drawn down to the bottom row (what
  CPR would tell prompt_toolkit), so the next count is exact when it shrank.
- Rows output scrolled while the width changed under it are repainted by the
  next refill (twice rather than truncated for good).
- A resize whose viewport holds the whole history erases from its oldest row,
  keeping the unrecorded startup banner.
- _terminal_reflows: TERM decides before inherited env (st/urxvt from tmux,
  kitty or vscode); rxvt added; vte-256color no longer matches 'vt'.
2026-09-24 00:17:13 -07:00

1877 lines
82 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Hermes Agent CLI — interactive terminal interface (``python cli.py --help`` for usage)."""
# Must be the very first import (UTF-8 stdio on Windows). Missing only mid-``hermes update``.
try:
import hermes_bootstrap # noqa: F401
except ModuleNotFoundError:
pass
import logging
import os
import functools
import shutil # noqa: F401 — tests patch shutil/time through the cli facade
import sys
import re
import atexit
import errno
import time # noqa: F401 — see shutil
from collections import deque
from dataclasses import dataclass
from contextlib import contextmanager, suppress
from pathlib import Path
from datetime import datetime # noqa: F401 — siblings import it lazily through cli
from typing import List, Dict, Any, Optional
logger = logging.getLogger(__name__)
os.environ["HERMES_QUIET"] = "1" # suppress our modules' startup chatter
from hermes_cli.cli_agent_setup_mixin import CLIAgentSetupMixin
from hermes_cli.cli_commands_mixin import CLICommandsMixin
from hermes_cli.cli_billing_mixin import CLIBillingMixin
from hermes_cli.cli_loops_mixin import CLILoopsMixin
from hermes_cli.cli_info_mixin import CLIInfoMixin
from hermes_cli.cli_terminal_mixin import CLITerminalMixin
from hermes_cli.cli_modal_mixin import CLIModalMixin
from hermes_cli.cli_stream_mixin import CLIStreamMixin
from hermes_cli.cli_session_mixin import CLISessionMixin
from hermes_cli.cli_model_switch_mixin import CLIModelSwitchMixin
from hermes_cli.cli_voice_mixin import CLIVoiceMixin
from hermes_cli.cli_status_bar_mixin import CLIStatusBarMixin
from hermes_cli.cli_tui_mixin import CLITuiMixin
from hermes_cli.cli_process_notifications import CLIProcessNotificationsMixin
from hermes_cli.cli_init_mixin import CLIInitMixin
from hermes_cli.cli_tui_runtime_mixin import CLITuiRuntimeMixin
# Extracted clusters (mechanical split, #116911); re-exported here so `cli.<name>` stays the seam.
from hermes_cli.cli_shutdown import ( # noqa: F401,E402
_CLEANUP_STEPS,
_arm_exit_watchdog,
_emit_interrupted_session_end,
_exit_watchdog_timeout,
_finalize_single_query,
_float_env,
_flush_logging_and_stdio,
_flush_one_shot_session_store,
_interrupt_async_delegations,
_invoke_interrupted_session_end,
_notify_session_finalize,
_notify_single_query_session_finalize,
_oneshot_agent_and_session,
_should_emit_cleanup_session_finalize,
_shutdown_agent_memory_provider,
_shutdown_cached_aux_clients,
_shutdown_mcp_servers,
_stop_cli_wake_word,
_sync_process_session_id,
_wait_for_oneshot_background_completions,
)
from hermes_cli.cli_auto_maintenance import ( # noqa: F401,E402
_run_checkpoint_auto_maintenance,
_run_state_db_auto_maintenance,
)
from hermes_cli.cli_render import ( # noqa: F401,E402
ChatConsole,
_ACCENT,
_ACCENT_ANSI_DEFAULT,
_BOLD,
_DA1_REPLY_RE,
_DIM,
_FALSE_RE,
_LIGHT_DEFAULT_TERM_PROGRAMS,
_LIGHT_MODE_REMAP,
_LIGHT_MODE_REMAP_UPPER,
_REASONING_TAGS,
_RST,
_STREAM_PAD,
_STREAM_PARTIAL_PREVIEW_LEN,
_SkinAwareAnsi,
_TOOL_CALL_TAGS,
_TRUE_RE,
_WINDOWS_PATH_WITH_DOT_SEGMENT_RE,
_accent_hex,
_add_suspect_rows,
_append_blank_panel_line,
_append_panel_line,
_assistant_content_as_text,
_assistant_copy_text,
_b,
_build_compact_banner,
_clear_output_history,
_cli_visible_print,
_coerce_output_history_limit,
_cprint,
_d,
_detect_light_mode_uncached,
_heal_cooked_mode_drift,
_hex_to_ansi,
_install_skin_light_mode_hook,
_line_rows,
_luminance_from_hex,
_maybe_remap_for_light_mode,
_output_history_recording,
_output_tail_fitting,
_painted_columns,
_PaintedLine,
_panel_box_width,
_post_stream_transform_output,
_prepend_note_to_message,
_preserve_windows_dot_segments_for_markdown,
_pt_app_is_running,
_pt_print_ansi,
_query_osc11_background,
_record_output_history,
_record_output_history_entry,
_release_paints,
_render_final_assistant_content,
_rich_text_from_ansi,
_set_chrome_floor,
_strip_markdown_syntax,
_strip_reasoning_tags,
_terminal_columns,
_terminal_reflows,
_terminal_width_for_streaming,
_tty_wrap,
_wrap_panel_text,
_wrap_panel_text_keep_ws,
)
from hermes_cli.cli_config_load import ( # noqa: F401,E402
_AUXILIARY_TASK_ENV,
_CWD_PLACEHOLDERS,
_TERMINAL_ENV_MAPPINGS,
_cli_config_defaults,
_init_logging_and_display_from_config,
_load_prefill_messages,
_merge_file_config,
_mirror_config_to_env,
_parse_reasoning_config,
_parse_service_tier_config,
_resolve_prefill_messages_file,
load_cli_config,
)
from hermes_cli.cli_terminal_input import ( # noqa: F401,E402
_BACKSLASH_LINE_CONTINUATION_RE,
_DSR_CPR_ESC_RE,
_DSR_CPR_VISIBLE_RE,
_EXTENDED_ENTER_KEYS_SEQ,
_IMAGE_EXTENSIONS,
_KITTY_KEYBOARD_PUSH_SEQ,
_MODIFY_OTHER_KEYS_SEQ,
_SGR_MOUSE_BARE_RE,
_SGR_MOUSE_ESC_RE,
_SGR_MOUSE_VISIBLE_RE,
_TERMINAL_INPUT_MODE_RESET_SEQ,
_apply_backslash_line_continuation,
_apply_bracketed_paste_timeout_patch,
_bind_prompt_submit_keys,
_build_cpr_disabled_output,
_cli_multiline_shortcuts_enabled,
_collect_query_images,
_detect_file_drop,
_disable_prompt_toolkit_cpr_warning,
_enable_extended_enter_keys,
_estimate_tui_input_height,
_file_drop_result,
_format_image_attachment_badges,
_hermes_call_output_screen_diff,
_is_backslash_line_continuation,
_is_ghostty_terminal,
_preserve_ctrl_enter_newline,
_resolve_attachment_path,
_select_classic_cli_pt_output,
_should_auto_attach_clipboard_image_on_paste,
_split_path_input,
_status_bar_visible_from_display_config,
_strip_leaked_terminal_responses_with_meta,
_terminal_may_leak_cpr,
_terminal_supports_extended_enter_keys,
_termux_example_image_path,
)
from hermes_cli.cli_single_query import ( # noqa: F401,E402
_TERMINAL_PROVIDER_REASONS,
_TRANSIENT_PROVIDER_REASONS,
_collect_kanban_task_images,
_configure_quiet_agent,
_install_single_query_signal_handlers,
_int_or,
_interrupt_agent_for_signal,
_route_single_query_images,
_run_kanban_goal_loop_chat,
_run_kanban_goal_loop_q,
_run_quiet_single_query,
_run_single_query_mode,
_single_query_exit_code,
_sync_cli_session_id_from_agent,
)
from prompt_toolkit.patch_stdout import patch_stdout
try:
from prompt_toolkit.enums import EditingMode
except ImportError: # partial prompt_toolkit stubs in tests
EditingMode = None
from prompt_toolkit import print_formatted_text as _pt_print
from prompt_toolkit.formatted_text import ANSI as _PT_ANSI
try:
from prompt_toolkit.cursor_shapes import CursorShape
_STEADY_CURSOR = CursorShape.BLOCK
except (ImportError, AttributeError):
_STEADY_CURSOR = None
try:
from hermes_cli import pt_input_extras as _pt_extras
_pt_extras.install_shift_enter_alias()
_pt_extras.install_ctrl_enter_alias()
_pt_extras.install_cmd_backspace_alias()
_pt_extras.install_modify_other_keys_aliases()
_pt_extras.install_keypress_data_normalization()
_pt_extras.install_ignored_terminal_sequences()
del _pt_extras
except Exception:
pass
import threading
import queue
def _lazy_shim(module: str, name: str, alias: str | None = None):
"""Import ``module.name`` on first call; keeps heavy imports off startup while ``cli.<name>`` stays patchable."""
import importlib
def shim(*args, **kwargs):
return getattr(importlib.import_module(module), name)(*args, **kwargs)
shim.__name__ = shim.__qualname__ = alias or name
return shim
def format_duration_compact(*args, **kwargs):
seconds = float(args[0] if args else kwargs.get("seconds", 0.0))
if seconds < 60:
return f"{seconds:.0f}s"
minutes = seconds / 60
if minutes < 60:
return f"{minutes:.0f}m"
hours = minutes / 60
if hours < 24:
remaining_min = int(minutes % 60)
return f"{int(hours)}h {remaining_min}m" if remaining_min else f"{int(hours)}h"
days = hours / 24
return f"{days:.1f}d"
# model id -> shortest configured alias (process-lifetime cache; config is read once).
_REVERSE_ALIAS_CACHE: dict[str, str] | None = None
def _reverse_alias_for_display(model_name: str) -> str:
"""Shortest alias for ``model_name`` from ``model_aliases:`` or ``model.aliases:``, else ``model_name``."""
global _REVERSE_ALIAS_CACHE
if not model_name:
return model_name
if _REVERSE_ALIAS_CACHE is None:
rmap: dict[str, str] = {}
def _put(m: str, alias: str) -> None:
if m and (m not in rmap or len(alias) < len(rmap[m])):
rmap[m] = alias
try:
from hermes_cli.config import load_config
cfg = load_config() or {}
ma = cfg.get("model_aliases")
if isinstance(ma, dict):
for alias, entry in ma.items():
if isinstance(entry, dict):
_put(str(entry.get("model", "") or "").strip(), alias)
mdl = cfg.get("model", {}) or {}
if isinstance(mdl, dict):
simple = mdl.get("aliases")
if isinstance(simple, dict):
for alias, val in simple.items():
if isinstance(val, str) and val.strip():
v = val.strip()
_put(v.split("/", 1)[1] if "/" in v else v, alias)
except Exception:
pass
_REVERSE_ALIAS_CACHE = rmap
return _REVERSE_ALIAS_CACHE.get(model_name, model_name)
def format_token_count_compact(*args, **kwargs):
value = int(args[0] if args else kwargs.get("value", 0))
abs_value = abs(value)
if abs_value < 1_000:
return str(value)
sign = "-" if value < 0 else ""
units = ((1_000_000_000, "B"), (1_000_000, "M"), (1_000, "K"))
for threshold, suffix in units:
if abs_value >= threshold:
scaled = abs_value / threshold
text = f"{scaled:.{2 if scaled < 10 else 1 if scaled < 100 else 0}f}"
if "." in text:
text = text.rstrip("0").rstrip(".")
return f"{sign}{text}{suffix}"
return f"{value:,}"
realign_markdown_tables = _lazy_shim("agent.markdown_tables", "realign_markdown_tables")
_COMMAND_SPINNER_FRAMES = ("⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏")
# ~/.hermes/.env first, project .env as dev fallback; user env files override stale shell exports.
from hermes_constants import get_hermes_home
from hermes_cli.env_loader import load_hermes_dotenv
_hermes_home = get_hermes_home()
_project_env = Path(__file__).parent / '.env'
load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)
CLI_CONFIG = load_cli_config()
_init_logging_and_display_from_config()
# Neuter AsyncHttpxClientWrapper.__del__ before any AsyncOpenAI client exists: it
# schedules aclose() on the running loop (prompt_toolkit's, during idle), closing
# transports bound to dead worker loops ("Event loop is closed" / "Press ENTER to
# continue..."). A meta_path finder patches ``openai._base_client`` at first import —
# eager import costs ~166ms/30MB cold, and the patch is guaranteed to land before
# instantiation. See ``agent.auxiliary_client.neuter_async_httpx_del``.
try:
import sys as _httpx_neuter_sys
import importlib.util as _httpx_neuter_imp_util
class _AsyncHttpxDelNeuter:
"""Patch ``AsyncHttpxClientWrapper.__del__`` to a no-op when ``openai._base_client`` loads."""
_armed = True
def find_spec(self, fullname, path=None, target=None):
if not self._armed or fullname != "openai._base_client":
return None
# Disarm before delegating so the recursive find_spec doesn't loop through us.
self._armed = False
try:
_httpx_neuter_sys.meta_path.remove(self)
except ValueError:
pass
spec = _httpx_neuter_imp_util.find_spec(fullname)
if spec is None or spec.loader is None:
return None
_orig_exec = spec.loader.exec_module
def _patched_exec(module):
_orig_exec(module)
try:
cls = getattr(module, "AsyncHttpxClientWrapper", None)
if cls is not None:
cls.__del__ = lambda self: None # type: ignore[assignment]
except Exception:
pass
spec.loader.exec_module = _patched_exec # type: ignore[method-assign]
return spec
_httpx_neuter_sys.meta_path.insert(0, _AsyncHttpxDelNeuter())
except Exception:
pass
# Agent/tool systems load lazily: bare startup only needs the prompt.
def get_tool_definitions(*args, **kwargs):
from hermes_cli.mcp_startup import wait_for_mcp_discovery
from model_tools import get_tool_definitions as _get_tool_definitions
wait_for_mcp_discovery()
return _get_tool_definitions(*args, **kwargs)
validate_toolset = _lazy_shim("toolsets", "validate_toolset")
_cleanup_all_terminals = _lazy_shim("tools.terminal_tool", "cleanup_all_environments", "_cleanup_all_terminals")
set_sudo_password_callback = _lazy_shim("tools.terminal_tool", "set_sudo_password_callback")
set_approval_callback = _lazy_shim("tools.terminal_tool", "set_approval_callback")
set_secret_capture_callback = _lazy_shim("tools.skills_tool", "set_secret_capture_callback")
_cleanup_all_browsers = _lazy_shim("tools.browser_tool_lifecycle", "_emergency_cleanup_all_sessions", "_cleanup_all_browsers")
_cleanup_done = False # _run_cleanup runs exactly once
_cleanup_in_progress = False
_cli_wake_owner = None
# One-shot finalization runs before process cleanup (plugins see the boundary while the
# agent is attached); atexit cleanup must not finalize those sessions again.
_single_query_finalize_attempted_session_ids: set[str | None] = set()
# /handoff sessions belong to the gateway: finalizing them here would stamp end_reason on
# a row the gateway just reopened, making the handoff leg vanish from history.
# Session IDs that were handed off to the gateway via /handoff. The CLI process exits after a successful
# handoff, but the gateway now owns the session lifecycle — _run_cleanup must NOT call finalize_session on
# these, because doing so sets end_reason on a row the gateway just reopened and is actively writing to
# (#88234). The race made the handoff leg vanish from session history and broke session_search recall for
# the handed-off session.
_handed_off_session_ids: set[str | None] = set()
_active_agent_ref = None # active AIAgent, for memory-provider shutdown at exit
_deferred_agent_startup_done = False
# Set once the TUI app starts (focus reporting + mouse tracking on); gates the on-exit
# terminal reset so non-TUI one-shot runs never emit codes for modes they never enabled.
_tui_input_modes_active = False
# Set True once the TUI's prompt_toolkit app starts (which enables focus reporting + mouse tracking). Gates
# the on-exit terminal reset so non-TUI one-shot CLI runs — which also register _run_cleanup via atexit —
# don't emit escape codes for modes they never enabled (#36823).
def _mark_tui_input_modes_active() -> None:
"""Record that the TUI app started, so _run_cleanup resets input modes."""
global _tui_input_modes_active
_tui_input_modes_active = True
def _prepare_deferred_agent_startup() -> None:
"""Run Termux-deferred agent discovery before the first real agent turn."""
global _deferred_agent_startup_done
if _deferred_agent_startup_done:
return
if os.environ.get("HERMES_DEFER_AGENT_STARTUP") != "1":
return
_deferred_agent_startup_done = True
_accept_hooks = os.environ.get("HERMES_ACCEPT_HOOKS", "").lower() in {"1", "true", "yes", "on"}
try:
from hermes_cli.plugins import discover_plugins
discover_plugins()
except Exception:
logger.warning("plugin discovery failed at deferred CLI startup", exc_info=True)
try:
from hermes_cli.mcp_startup import start_background_mcp_discovery
start_background_mcp_discovery(logger=logger, thread_name="termux-cli-mcp-discovery")
except Exception:
logger.debug("MCP tool discovery failed at deferred CLI startup", exc_info=True)
try:
from agent.shell_hooks import register_from_config
from agent.outbound_webhooks import register_from_config as register_outbound_webhooks
from hermes_cli.config import load_config
_hooks_cfg = load_config()
register_from_config(_hooks_cfg, accept_hooks=_accept_hooks)
register_outbound_webhooks(_hooks_cfg)
except Exception:
logger.debug("shell-hook registration failed at deferred CLI startup", exc_info=True)
_signal_watchdog_armed = False
def _arm_exit_watchdog_on_shutdown_signal() -> None:
"""Arm the exit backstop the moment a termination signal arrives (idempotent; never raises).
The graceful unwind has wedge points BEFORE ``_run_cleanup`` arms its own watchdog
(main thread in a syscall, prompt_toolkit teardown never returning). Leash is 2x
the cleanup timeout so a progressing cleanup is never cut short. Never arm at
startup: the timer exits unconditionally.
SIGTERM/SIGHUP establish unambiguous shutdown intent, but the graceful path from signal →
``agent.interrupt()`` → ``app.exit()`` / ``KeyboardInterrupt`` → ``finally`` → ``_run_cleanup`` has
several wedge points BEFORE ``_run_cleanup`` arms the normal watchdog: a main thread parked in a syscall
that never observes the unwind, a prompt_toolkit teardown that never returns, or an agent worker
blocking the ``finally``. When that happens the process has NO backstop and a "dead" CLI lingers
(observed: ``hermes --tui`` alive ~47 min at 4% CPU after terminal close — the #65998 class).
"""
global _signal_watchdog_armed
if _signal_watchdog_armed:
return
_signal_watchdog_armed = True
base = _exit_watchdog_timeout()
if base <= 0:
return # explicitly disabled
with suppress(Exception): # never let the backstop break signal handling
_arm_exit_watchdog(timeout_s=base * 2, from_signal=True)
def _run_cleanup(*, notify_session_finalize: bool = True):
"""Run resource cleanup exactly once."""
global _cleanup_done, _cleanup_in_progress
if _cleanup_done:
return
_cleanup_done = True
_cleanup_in_progress = True
try:
_arm_exit_watchdog()
# Reset terminal input modes FIRST: teardown below can take seconds and a later
# step raising must not skip the reset. No-op unless the TUI ran.
# See #36823.
_reset_terminal_input_modes_on_exit()
for step, swallow in _CLEANUP_STEPS:
with suppress(swallow):
globals()[step]()
if notify_session_finalize:
cleanup_session_id = _active_agent_ref.session_id if _active_agent_ref else None
if _should_emit_cleanup_session_finalize(cleanup_session_id):
_notify_session_finalize(session_id=cleanup_session_id, platform="cli", reason="shutdown")
try:
_shutdown_agent_memory_provider(_active_agent_ref)
except Exception as e:
logger.warning("CLI cleanup memory shutdown failed: %s", e, exc_info=True)
finally:
_cleanup_in_progress = False
def _reset_terminal_input_modes_on_exit() -> None:
"""Disable focus reporting + mouse tracking on TUI exit (best-effort).
Ctrl+C / SIGTERM / crashes bypass prompt_toolkit's unwind, leaving focus events and
mouse reports as visible text in the next shell. Writes to stdout when it is the
terminal, else /dev/tty (the TUI may have run with stdout redirected).
Called from ``_run_cleanup`` (atexit-registered + invoked on the normal / EOF / interrupt exit paths)
this covers normal quit, Ctrl+C and SIGTERM/SIGHUP. ``kill -9`` is uncatchable, and the kanban worker's
``os._exit(0)`` path bypasses ``atexit``; neither runs this — but both are non-TTY / non-TUI, so there
is nothing to reset there. See #36823.
"""
global _tui_input_modes_active
if not _tui_input_modes_active:
return
# Clear first so a re-armed _run_cleanup doesn't re-emit.
_tui_input_modes_active = False
try:
stream = sys.stdout
if stream is not None and stream.isatty():
stream.write(_TERMINAL_INPUT_MODE_RESET_SEQ)
stream.flush()
return
except Exception:
pass
with suppress(Exception), open("/dev/tty", "w", encoding="ascii") as tty:
tty.write(_TERMINAL_INPUT_MODE_RESET_SEQ)
tty.flush()
from hermes_cli.worktree_ops import (
_git_quiet,
_git_repo_root,
_maintain_pack_health,
_prune_stale_worktrees,
_repo_is_shallow,
_setup_worktree,
_worktree_has_unpushed_commits,
release_lsp_clients,
)
# ============================================================================= Git Worktree Isolation
# (#652) =============================================================================
_active_worktree: Optional[Dict[str, str]] = None
def _cleanup_worktree(info: Dict[str, str] = None) -> None:
"""Remove a worktree and its branch on exit; kept only when it has unpushed commits."""
global _active_worktree
info = info or _active_worktree
if not info:
return
wt_path, branch, repo_root = info["path"], info["branch"], info["repo_root"]
if not Path(wt_path).exists():
return
if _worktree_has_unpushed_commits(wt_path, timeout=10):
if _repo_is_shallow(repo_root):
# Shallow boundary makes the unpushed verdict unreliable; the startup pruner reaps later.
_cprint(f"\n\033[33m⚠ Shallow clone — cannot verify push state, keeping: {wt_path}\033[0m")
print(" The next `hermes -w` session deepens the clone and prunes merged worktrees automatically.")
else:
_cprint(f"\n\033[33m⚠ Worktree has unpushed commits, keeping: {wt_path}\033[0m")
print(f" To clean up manually: git worktree remove --force {wt_path}")
_active_worktree = None
return
# Release the tree's language servers while the path still exists, then unlock so `remove`
# isn't blocked by the lock placed at creation. Fail-soft.
release_lsp_clients(wt_path)
_git_quiet(["worktree", "unlock", wt_path], repo_root, log="git worktree unlock failed (non-fatal)")
_git_quiet(["worktree", "remove", wt_path, "--force"], repo_root, timeout=15, log="Failed to remove worktree")
_git_quiet(["branch", "-D", branch], repo_root, log=f"Failed to delete branch {branch}")
_active_worktree = None
_cprint(f"\033[32m✓ Worktree cleaned up: {wt_path}\033[0m")
# Light/dark terminal detection (mirrors ui-tui/src/theme.ts detectLightMode()). Priority:
# HERMES_LIGHT/HERMES_TUI_LIGHT env, HERMES_TUI_THEME, HERMES_TUI_BACKGROUND, COLORFGBG
# (bg slot 7/15 = light), OSC 11 query, default dark. Cached so the terminal is queried once.
_LIGHT_MODE_CACHE: bool | None = None
def _detect_light_mode() -> bool:
global _LIGHT_MODE_CACHE
if _LIGHT_MODE_CACHE is not None:
return _LIGHT_MODE_CACHE
try:
result = _detect_light_mode_uncached()
except Exception:
result = False
_LIGHT_MODE_CACHE = result
return result
_install_skin_light_mode_hook()
# Prime the light-mode cache when interactive so OSC 11 happens before prompt_toolkit owns the tty.
with suppress(Exception):
if sys.stdin.isatty() and sys.stdout.isatty():
_detect_light_mode()
_OUTPUT_HISTORY_ENABLED = True
_OUTPUT_HISTORY_REPLAYING = False
_OUTPUT_HISTORY_SUPPRESSED = False
_OUTPUT_HISTORY_MAX_LINES = 200
_OUTPUT_HISTORY = deque(maxlen=_OUTPUT_HISTORY_MAX_LINES)
def _configure_output_history(enabled: bool, max_lines=200) -> None:
"""Configure recent CLI output replayed after terminal redraws."""
global _OUTPUT_HISTORY_ENABLED, _OUTPUT_HISTORY_MAX_LINES, _OUTPUT_HISTORY
_OUTPUT_HISTORY_ENABLED = bool(enabled)
_OUTPUT_HISTORY_MAX_LINES = _coerce_output_history_limit(max_lines)
_OUTPUT_HISTORY = deque(maxlen=_OUTPUT_HISTORY_MAX_LINES)
@contextmanager
def _suspend_output_history():
global _OUTPUT_HISTORY_SUPPRESSED
old_value = _OUTPUT_HISTORY_SUPPRESSED
_OUTPUT_HISTORY_SUPPRESSED = True
try:
yield
finally:
_OUTPUT_HISTORY_SUPPRESSED = old_value
def _output_history_lines() -> list[str]:
"""The recorded output as the lines a replay paints (callable entries render now)."""
rendered_lines = []
for entry in tuple(_OUTPUT_HISTORY):
lines = [entry]
if callable(entry):
try:
lines = entry()
except Exception:
continue
if isinstance(lines, str):
lines = lines.splitlines()
rendered_lines.extend(line if isinstance(line, str) else str(line) for line in lines)
return rendered_lines
def _output_history_rows(limit: int, columns: int, painted: bool):
"""Rows the whole recorded output fills (counted as ``_output_tail_fitting`` does), or
``None`` when that is ``limit`` rows or more."""
if not _OUTPUT_HISTORY_ENABLED:
return None
total = 0
for line in reversed(_output_history_lines()):
total += _line_rows(line, (getattr(line, "width", None) if painted else None) or columns)
if total >= limit:
return None
return total
def _replay_output_history(fit=None, output=None) -> None:
"""Repaint recent output above the prompt after a full screen clear.
``fit=(rows, columns, painted, top)`` replays only the newest lines whose wrapped height
fits ``rows`` (see ``_output_tail_fitting``) — the older ones are still in scrollback
(#95375) — from screen row ``top`` when known (``_set_chrome_floor``). ``output``: paint
now, straight to this prompt_toolkit output, where the caller just erased the viewport and
reset the renderer — ``run_in_terminal`` would first erase below the top row, which
scroll-on-clear terminals (tmux) take as a clear and copy the blank screen into scrollback.
"""
global _OUTPUT_HISTORY_REPLAYING
if not _OUTPUT_HISTORY_ENABLED or not _OUTPUT_HISTORY:
return
_OUTPUT_HISTORY_REPLAYING = True
try:
rendered_lines = _output_history_lines()
top = None
if fit is not None:
rows, columns, painted, top = fit
rendered_lines = _output_tail_fitting(rendered_lines, rows, columns, painted)
if rendered_lines:
# One payload: per-line pt prints each force a sync redraw (a waterfall of old output).
if output is None:
_pt_print(_PT_ANSI("\n".join(rendered_lines)))
else:
from prompt_toolkit.renderer import print_formatted_text as _paint_formatted_text
from prompt_toolkit.styles import Style
_paint_formatted_text(output, _PT_ANSI("\n".join(rendered_lines) + "\n"), Style([]))
size = output.get_size()
if top is not None: # the chrome's top is now this many rows down
top += sum(_line_rows(line, columns) for line in rendered_lines)
_set_chrome_floor(max(0, size.rows - top))
if size.columns != columns:
_add_suspect_rows(top + 1 - size.rows)
width = _painted_columns() if fit is None else columns
for line in rendered_lines: # repainted: they wrap at today's width from now on
if isinstance(line, _PaintedLine):
line.width = width
except Exception:
pass
finally:
_OUTPUT_HISTORY_REPLAYING = False
_strip_leaked_bracketed_paste_wrappers = _lazy_shim(
"hermes_cli.input_sanitize", "strip_leaked_bracketed_paste_wrappers", "_strip_leaked_bracketed_paste_wrappers"
)
# OSC sequences (e.g. OSC-8 links): pt's ANSI parser strips the ESC but leaks the payload as text.
_OSC_ESCAPE_RE = re.compile(r"\x1b\][\s\S]*?(?:\x07|\x1b\\)")
def _looks_like_slash_command(text: str) -> bool:
"""``/help`` yes, ``/Users/x/file.md`` no: a command's first word has no further ``/``."""
if not text or not text.startswith("/"):
return False
return "/" not in text.split()[0][1:]
_skill_commands = None
_skill_bundles = None
def _slash_args(cmd: str) -> str:
"""Text after the slash-command word, stripped ("" when absent)."""
parts = cmd.split(None, 1)
return parts[1].strip() if len(parts) > 1 else ""
def _ensure_skill_commands() -> dict:
global _skill_commands
if _skill_commands is None:
from agent.skill_commands import scan_skill_commands
_skill_commands = scan_skill_commands()
return _skill_commands
def get_skill_commands() -> dict:
return _ensure_skill_commands()
build_skill_invocation_message = _lazy_shim("agent.skill_commands", "build_skill_invocation_message")
build_preloaded_skills_prompt = _lazy_shim("agent.skill_commands", "build_preloaded_skills_prompt")
def get_skill_bundles() -> dict:
global _skill_bundles
if _skill_bundles is None:
from agent.skill_bundles import get_skill_bundles as _impl
_skill_bundles = _impl()
return _skill_bundles
build_bundle_invocation_message = _lazy_shim("agent.skill_bundles", "build_bundle_invocation_message")
def _get_plugin_cmd_handler_names() -> set:
"""Return plugin command names (without slash prefix) for dispatch matching."""
try:
from hermes_cli.plugins import get_plugin_commands
return set(get_plugin_commands().keys())
except Exception:
return set()
def _parse_skills_argument(skills: str | list[str] | tuple[str, ...] | None) -> list[str]:
"""Normalize a CLI skills flag into a deduplicated list of skill identifiers."""
if not skills:
return []
raw_values = [str(item) for item in skills if item is not None] if isinstance(skills, (list, tuple)) else [str(skills)]
parts = (p.strip() for raw in raw_values for p in raw.split(","))
return list(dict.fromkeys(p for p in parts if p))
def save_config_value(key_path: str, value: any) -> bool:
"""Persist dot-separated ``key_path`` = value into HERMES_HOME/config.yaml; True on success.
Never the repo's cli-config.yaml: no config reader loads it, so the value would vanish.
"""
config_path = get_hermes_home() / 'config.yaml'
try:
from hermes_constants import mkdir_under_hermes_home
mkdir_under_hermes_home(config_path.parent)
from utils import atomic_roundtrip_yaml_update
atomic_roundtrip_yaml_update(config_path, key_path, value)
try: # owner-only: config files contain API keys
os.chmod(config_path, 0o600)
except (OSError, NotImplementedError):
pass
return True
except Exception as e:
logger.error("Failed to save config: %s", e)
return False
def _normalize_moa_model(model: Optional[str]) -> tuple[Optional[str], Optional[str]]:
"""``moa:<preset>`` -> ``("moa", preset)`` (same routing as ``/moa``); anything else -> ``(None, model)``.
Returns ``("moa", "<preset>")`` when *model* selects the MoA virtual provider, otherwise ``(None,
model)`` unchanged. This gives non-interactive ``hermes chat -Q -m moa:<preset>`` the same routing the
interactive ``/moa`` command and the model picker already use: ``resolve_runtime_provider`` handles
``requested_provider == "moa"`` and ``agent_init`` builds the MoAClient off ``provider == "moa"``.
Without this the raw ``moa:<preset>`` string is sent to the real provider and rejected with a 401/400
"model not supported" (#56828).
"""
if isinstance(model, str) and model.strip().lower().startswith("moa:"):
preset = model.strip().split(":", 1)[1].strip()
if preset:
return "moa", preset
return None, model
_split_model_config_default = _lazy_shim("hermes_cli.config", "split_model_config_default", "_split_model_config_default")
class _VoiceInputMessage:
"""Sentinel for voice-transcribed input so the concise voice prefix never applies to typed text.
Distinguishes STT output from manually typed text while voice mode is active, so the
concise-voice-response prefix is applied only to messages that actually came from the microphone
(#65827).
"""
__slots__ = ("text",)
def __init__(self, text: str):
self.text = text
def __str__(self) -> str:
return self.text
class _SeededQueryMessage:
"""Sentinel for a ``-q`` prompt seeded into an interactive session; treated LITERALLY (no slash/!/file-drop)."""
__slots__ = ("text", "images")
def __init__(self, text: str, images=None):
self.text = text or ""
self.images = list(images or [])
def __str__(self) -> str:
return self.text
def _should_seed_interactive(query, image, quiet: bool, oneshot: bool) -> bool:
"""``-q`` seeds an interactive session only on a real TTY without ``--oneshot``/``-Q`` (automation answers and exits)."""
if not (query or image) or oneshot or quiet:
return False
try:
return bool(sys.stdin.isatty() and sys.stdout.isatty())
except Exception:
return False
@dataclass
class _ChatTurn:
"""Per-turn state shared by the ``chat()`` phases and the agent worker thread.
``result`` is written by the worker and read after the join; ``tts_normal_exit`` is
set only when the TTS worker drained on its own so the last sentence is never cut.
"""
result: Optional[dict] = None
mute_notification_reply: bool = False
use_streaming_tts: bool = False
box_opened: bool = False
thinking_started: bool = False
text_queue: Optional[queue.Queue] = None
tts_thread: Optional[threading.Thread] = None
stream_callback: Optional[Any] = None
stop_event: Optional[threading.Event] = None
tts_normal_exit: bool = False
voice_prefix: str = ""
from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
_PASTE_REF_RE = re.compile(r'\[Pasted text #\d+: \d+ lines \u2192 (.+?)\]')
class HermesCLI(CLIInitMixin, CLITuiRuntimeMixin, CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMixin, CLIStatusBarMixin, CLIVoiceMixin, CLIModelSwitchMixin, CLISessionMixin, CLIStreamMixin, CLIModalMixin, CLITerminalMixin, CLIInfoMixin, CLILoopsMixin, CLIChatTurnMixin):
"""Interactive REPL for the Hermes Agent."""
# Seeded -q first message (see _should_seed_interactive); run() re-creates
# _pending_input, so it is enqueued only after the fresh queue exists.
_seeded_first_message: Optional["_SeededQueryMessage"] = None
# Inspection surfaces (banner, /tools, status line) read this on partially built instances too.
disabled_toolsets: Optional[List[str]] = None
def __init__(
self,
model: str = None,
toolsets: List[str] = None,
provider: str = None,
reasoning: str = None,
api_key: str = None,
base_url: str = None,
max_turns: int = None,
run_budget: float = None,
verbose: Optional[bool] = None,
compact: bool = False,
resume: str = None,
checkpoints: bool = False,
pass_session_id: bool = False,
ignore_rules: bool = False,
):
"""CLI args win over config; ``reasoning`` is per-run only; ``resume`` restores history from SQLite."""
self._init_display_options(verbose, compact)
self._init_model_routing(model, toolsets, provider, reasoning, api_key, base_url, max_turns, run_budget,
checkpoints, pass_session_id, ignore_rules)
self._init_runtime_state(resume)
def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool:
"""Claim a global active-session slot for this CLI process."""
if self._active_session_lease is not None:
return True
try:
from hermes_cli.active_sessions import format_refusal_stderr, try_acquire_active_session
lease, message = try_acquire_active_session(
session_id=self.session_id,
surface=surface,
config=self.config,
# Writer identity: a re-claim by this process replaces its own entry.
# See #94595.
metadata={"live_session_id": str(self.session_id)},
)
except Exception as exc:
logger.warning("Failed to claim active session slot: %s", exc)
return True
if message:
print(format_refusal_stderr(message), file=sys.stderr) if stderr else self._console_print(f"[bold red]{message}[/]")
return False
self._active_session_lease = lease
with suppress(Exception):
atexit.register(self._release_active_session)
return True
def _release_active_session(self) -> None:
lease = getattr(self, "_active_session_lease", None)
if lease is None:
return
try:
lease.release()
except Exception:
logger.debug("Failed to release active session slot", exc_info=True)
finally:
self._active_session_lease = None
_PET_FRAME_INTERVAL = 0.16
_PET_CFG_INTERVAL = 2.5
def _install_tool_callbacks(self) -> None:
"""Install tool callbacks that need the live prompt UI."""
if self._tool_callbacks_installed:
return
set_sudo_password_callback(self._sudo_password_callback)
set_approval_callback(self._approval_callback)
set_secret_capture_callback(self._secret_capture_callback)
from agent.vault_backends.unlock import set_code_prompt_callback, set_save_login_prompt_callback, set_unlock_prompt_callback
set_unlock_prompt_callback(self._vault_unlock_callback)
set_save_login_prompt_callback(self._vault_save_login_callback)
set_code_prompt_callback(self._vault_code_callback)
self._tool_callbacks_installed = True
def _ensure_tirith_security(self) -> None:
"""Check tirith availability once before tools can run terminal commands."""
if self._tirith_security_checked:
return
self._tirith_security_checked = True
try:
from tools.tirith_security import ensure_installed, is_platform_supported
if (
ensure_installed(log_failures=False) is None and is_platform_supported()
and (self.config.get("security", {}) or {}).get("tirith_enabled", True)
):
_cprint(
f" {_DIM}⚠ tirith security scanner enabled but not available "
f"— command scanning will use pattern matching only{_RST}"
)
except Exception:
pass
def _show_security_advisories(self):
"""Startup banner for unacked security advisories, on stderr (piped stdout stays clean); 24h rate-limited."""
try:
from hermes_cli.security_advisories import detect_compromised, startup_banner
banner = startup_banner(detect_compromised())
if banner:
print(banner, file=sys.stderr, flush=True)
except Exception:
pass # never block startup
def _show_browser_backend_notice(self):
"""Once-per-24h hint when the default Browser Use backend silently fell back to built-in tools."""
try:
from tools.browser_use_cli import default_downgrade_notice
notice = default_downgrade_notice()
if notice:
from gateway.warning_notifications import render_notification
render_notification(lambda: self._console_print(f"[yellow]⚠ {notice}[/yellow]"), platform="cli")
except Exception:
logger.debug("browser backend notice failed", exc_info=True)
def finalize_preloaded_skills(self) -> None:
"""Join the background --skills preload and fold it into the prompt (idempotent).
Raises ``ValueError`` only when EVERY requested skill was unknown.
"""
if getattr(self, "_preload_skills_finalized", False):
return
thread = getattr(self, "_preload_skills_thread", None)
if thread is None:
self._preload_skills_finalized = True
return
thread.join(timeout=120)
self._preload_skills_finalized = True
err = getattr(self, "_preload_skills_error", None)
if err is not None:
raise err
auto_result = getattr(self, "_auto_load_skills_result", None)
if auto_result and auto_result[2]:
logger.warning("skills.auto_load: skill(s) not found or disabled, skipped: %s", ", ".join(auto_result[2]))
# auto_load names first, then explicit -s names that were not already pinned.
self.preloaded_skills = list(auto_result[1]) if auto_result else []
result = getattr(self, "_preload_skills_result", None)
if not result:
return
skills_prompt, loaded_skills, missing_skills = result
if missing_skills:
missing_display = ", ".join(missing_skills)
# A typo'd name must not crash a kanban worker; only a fully-missing set fails loudly.
if loaded_skills:
logger.warning(
"Unknown skill(s) requested, skipping: %s. "
"Continuing with: %s. "
"List available skills with `hermes skills list`.",
missing_display,
", ".join(loaded_skills),
)
else:
raise ValueError(f"Unknown skill(s): {missing_display}")
if skills_prompt:
self.system_prompt = "\n\n".join(p for p in (self.system_prompt, skills_prompt) if p).strip()
self.preloaded_skills += [name for name in loaded_skills if name not in self.preloaded_skills]
def _show_tool_availability_warnings(self):
"""Warn about toolsets switched off at startup (missing API keys, unusable terminal backend)."""
try:
# Runs on a daemon thread on the snapshot fast path: keep the imports to modules the
# registry walk already loaded plus the pure notices module (a heavy import here races
# importlib's module locks against the main thread).
from model_tools import check_tool_availability
from hermes_cli.tool_availability_notices import (
current_terminal_backend, filter_to_enabled_toolsets, tool_availability_warning_lines,
)
from tools.terminal_tool import terminal_backend_unavailable_reason
from toolsets import resolve_toolset
_, unavailable = check_tool_availability()
# Only toolsets this CLI session actually has. The selection is usually a composite bundle
# (``hermes-cli``), so expand it to tool names before matching — a raw name comparison
# matched nothing on a default install and silently dropped the terminal notice.
unavailable = filter_to_enabled_toolsets(unavailable, self.enabled_toolsets or [], resolve_toolset)
lines = tool_availability_warning_lines(
unavailable, terminal_reason=terminal_backend_unavailable_reason(),
terminal_backend=current_terminal_backend())
if lines:
self._console_print()
for line in lines:
self._console_print(line)
except Exception:
pass
def show_config(self):
"""Display current configuration with kawaii ASCII art."""
terminal_env = os.getenv("TERMINAL_ENV", "local")
terminal_cwd = os.getenv("TERMINAL_CWD", os.getcwd())
terminal_timeout = os.getenv("TERMINAL_TIMEOUT", "60")
config_path = _hermes_home / 'config.yaml'
if not config_path.exists():
config_path = Path(__file__).parent / 'cli-config.yaml'
config_status = "(loaded)" if config_path.exists() else "(not found)"
# ``api_key`` may be a callable (Entra ID bearer provider): never invoke it. Prefer the
# LIVE agent's key: the constructor seeds self.api_key from env before provider
# resolution, so on non-OpenAI providers it can be another vendor's key.
from agent.azure_identity_adapter import is_token_provider
display_key = self.api_key
if self.agent is not None and getattr(self.agent, "api_key", None):
display_key = self.agent.api_key
if is_token_provider(display_key):
api_key_display = "Microsoft Entra ID"
elif isinstance(display_key, str) and len(display_key) > 12:
api_key_display = f"{display_key[:8]}...{display_key[-4:]}"
else:
api_key_display = "Not set!"
title = "(^_^) Configuration"
width = 50
pad = width - len(title)
ssh_target = (
f"{os.getenv('TERMINAL_SSH_USER', 'not set')}@{os.getenv('TERMINAL_SSH_HOST', 'not set')}"
f":{os.getenv('TERMINAL_SSH_PORT', '22')}"
) if terminal_env == "ssh" else None
sections = (
("Model", (("Model: ", self.model), ("Base URL: ", self.base_url), ("API Key: ", api_key_display))),
("Terminal", (
("Environment: ", terminal_env),
*((("SSH Target: ", ssh_target),) if ssh_target else ()),
("Working Dir: ", terminal_cwd),
("Timeout: ", f"{terminal_timeout}s"),
)),
("Agent", (
("Max Turns: ", self.max_turns),
("Toolsets: ", ", ".join(self.enabled_toolsets) if self.enabled_toolsets else "all"),
("Verbose: ", self.verbose),
)),
("Session", (
("Started: ", self.session_start.strftime("%Y-%m-%d %H:%M:%S")),
("Config File:", f"{config_path} {config_status}"),
)),
)
print()
print("+" + "-" * width + "+")
print("|" + " " * (pad // 2) + title + " " * (pad - pad // 2) + "|")
print("+" + "-" * width + "+")
for name, rows in sections:
print()
print(f" -- {name} --")
for label, value in rows:
print(f" {label} {value}")
print()
# canonical command -> (method name, pass cmd_original?). Absent commands resolve to
# ``_handle_<name>_command(cmd)``. Looked up via getattr at dispatch time so
# monkeypatching works. A handler returning False exits the REPL.
_SLASH_DISPATCH: dict[str, tuple[str, bool]] = {
"exit": ("_cmd_exit", True), "quit": ("_cmd_exit", True), "help": ("_cmd_help", True),
"palette": ("_open_command_palette", False), "whoami": ("_handle_whoami_command", False),
"profile": ("_handle_profile_command", False), "toolsets": ("show_toolsets", False),
"config": ("show_config", False), "redraw": ("_cmd_redraw", True), "clear": ("_cmd_clear", True),
"history": ("show_history", False), "title": ("_cmd_title", True), "new": ("_cmd_new", True),
"model": ("_handle_model_switch", True), "codex-runtime": ("_handle_codex_runtime", True),
"retry": ("_cmd_retry", True), "prompt": ("_handle_prompt_compose_command", True),
"undo": ("_cmd_undo", True), "save": ("save_conversation", True), "skills": ("_cmd_skills", True),
"platforms": ("_show_gateway_status", False), "status": ("_show_session_status", False),
"context": ("_show_context_breakdown", True), "egress": ("_cmd_egress", True),
"statusbar": ("_cmd_statusbar", True), "verbose": ("_toggle_verbose", False), "yolo": ("_toggle_yolo", False),
"compress": ("_manual_compress", True), "subscription": ("_show_subscription", False),
"topup": ("_show_billing", True), "insights": ("_show_insights", True), "update": ("_cmd_update", True),
"version": ("_cmd_version", True), "paste": ("_handle_paste_command", False), "reload": ("_cmd_reload", True),
"reload-mcp": ("_confirm_and_reload_mcp", True), "reload-skills": ("_cmd_reload_skills", True),
"plugins": ("_cmd_plugins", True), "stop": ("_handle_stop_command", False),
"agents": ("_handle_agents_command", False), "bg": ("_handle_background_command", True),
"queue": ("_cmd_queue", True), "steer": ("_cmd_steer", True), "moa": ("_cmd_moa", True),
}
@classmethod
def _slash_handler(cls, canonical: str) -> tuple[str, bool] | None:
"""(method name, pass cmd_original?) for a registered command, else None."""
entry = cls._SLASH_DISPATCH.get(canonical)
if entry is None:
name = f"_handle_{canonical.replace('-', '_')}_command"
if callable(getattr(cls, name, None)):
entry = (name, True)
return entry
def process_command(self, command: str) -> bool:
"""Dispatch a slash command; returns False to exit the REPL."""
cmd_lower = command.lower().strip() # lowercase only for matching; args keep their case
cmd_original = command.strip()
# Aliases resolve via the central registry (hermes_cli/commands.py).
from hermes_cli.commands import resolve_command as _resolve_cmd
_base_word = cmd_lower.split()[0].lstrip("/")
_cmd_def = _resolve_cmd(_base_word)
canonical = _cmd_def.name if _cmd_def else _base_word
# Observer-only pre_command plugin hook (return values ignored; never raises).
if _cmd_def is not None:
from hermes_cli.plugins import fire_pre_command_hook
fire_pre_command_hook(
surface="cli", command=canonical, alias_used=_base_word, args_raw=_slash_args(cmd_original),
session_key=getattr(self, "session_id", None), platform="cli",
)
# A bare `/resume` prompt is one-shot: any other command disarms it so a later
# number isn't swallowed as a stale selection.
# See #34584.
if canonical not in {"resume", "sessions"}:
# Armed when a bare `/resume` prints the recent-sessions list so the very next bare numeric
# input (e.g. `3`) resolves to that session. Holds the exact list used for index resolution;
# one-shot (cleared on the next submitted input, whether it's the selection or anything else).
# See #34584.
self._pending_resume_sessions = None
entry = self._slash_handler(canonical)
if entry is None:
return self._process_unregistered_slash(cmd_original, cmd_lower)
method_name, pass_arg = entry
handler = getattr(self, method_name)
result = handler(cmd_original) if pass_arg else handler()
return result is not False
def _process_unregistered_slash(self, cmd_original: str, cmd_lower: str) -> bool:
"""Slash input with no built-in handler; precedence: quick_commands -> plugins -> bundles -> skills -> prefix expansion."""
base_cmd = cmd_lower.split()[0]
bare = base_cmd.lstrip("/")
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 bare in quick_commands:
return self._run_quick_command(base_cmd, quick_commands[bare], user_args)
if bare in _get_plugin_cmd_handler_names():
self._run_plugin_slash_command(base_cmd, user_args)
elif base_cmd in skill_bundles:
self._run_skill_bundle_command(base_cmd, skill_bundles[base_cmd], user_args)
elif base_cmd in skill_commands:
self._run_skill_slash_command(base_cmd, skill_commands[base_cmd], user_args)
else:
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 == "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[/]")
return True
if qtype != "exec":
self._console_print(f"[bold red]Quick command '{base_cmd}' has unsupported type (supported: 'exec', 'alias')[/]")
return True
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-authored config snippets, never LLM controlled);
# the env is sanitized because this process holds every API key.
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(),
creationflags=windows_hide_flags(), # no console flash on Windows (#56747)
)
# See #56747.
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:
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}[/]")
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 not plugin_handler:
return
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:
"""``/<bundle>`` 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
self._queue_loaded_skills(msg, f"Loading bundle: {bundle_info['name']} ({len(loaded_names)} skills)", missing)
def _queue_loaded_skills(self, msg, label: str, missing) -> None:
print(f"\n⚡ {label}")
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:
"""``/<skill> ...``; 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
self._queue_loaded_skills(
msg, f"Loading {len(loaded_names)} stacked skills: {', '.join(loaded_names)}", missing
)
return
msg = build_skill_invocation_message(base_cmd, rest, task_id=self.session_id)
if msg:
self._queue_loaded_skills(msg, f"Loading skill: {skill_info['name']}", None)
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:
if typed_base in matches:
matches = [typed_base]
else:
# 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.
from hermes_cli.cli_unknown_command import unknown_command_lines
lead, pointer = unknown_command_lines(cmd_lower, all_known)
_cprint(f"\033[1;31m{lead}{_RST}")
_cprint(f"{_DIM}{_ACCENT}{pointer}{_RST}")
return True
def _drain_interrupt_queue_to_pending_input(self) -> None:
"""Move stray ``_interrupt_queue`` messages into ``_pending_input`` after every turn.
Busy-time input lands in ``_interrupt_queue`` and is only drained by the explicit
interrupt path; a turn that finishes naturally would otherwise strand it and the
CLI appears to hang. Never raises.
Called once at the end of every turn from ``process_loop``'s ``finally`` block. Catches and swallows
``Exception`` because the drain must never break the main loop. (#20271)
"""
try:
while not self._interrupt_queue.empty():
stray = self._interrupt_queue.get_nowait()
if stray:
self._pending_input.put(stray)
except Exception:
pass
def _on_reasoning(self, reasoning_text: str):
"""Callback for intermediate reasoning display during tool-call loops."""
if not reasoning_text:
return
self._reasoning_preview_buf = getattr(self, "_reasoning_preview_buf", "") + reasoning_text
self._flush_reasoning_preview(force=False)
# Inline tokens that bypass the destructive-slash confirmation modal (scripting, or
# when the modal can't be marshaled onto the app loop).
# A general escape hatch for non-interactive use (scripting/automation) and for the degraded path where
# the modal can't be marshaled onto the app loop — lets users self-serve without flipping
# approvals.destructive_slash_confirm in config. (Native Windows now drives the modal normally — see
# #33961.)
_DESTRUCTIVE_SKIP_TOKENS = frozenset({"now", "--yes", "-y"})
def run(self):
"""Run the interactive CLI loop with persistent input at bottom."""
if not self._claim_active_session("cli"):
return
self._tui_print_startup()
self._tui_init_run_state()
kb = self._tui_build_key_bindings()
layout, style = self._tui_build_layout(kb)
app = self._tui_build_application(layout, kb, style)
_disable_prompt_toolkit_cpr_warning(app)
app.after_render += self._pet_flush_kitty_frame
self._app = app
# Ghost status-bar lines on resize: pt's renderer scrolls the terminal after each
# paint, pushing chrome into scrollback where a column-shrink reflows it into
# duplicates. Wrapping _output_screen_diff keeps its reserve-space branch from firing.
try:
# Background: prompt_toolkit's renderer (renderer.py L232-242) explicitly moves the cursor to
# the bottom of the canvas after painting "to make sure the terminal scrolls up, even when the
# lower lines of the canvas just contain whitespace". In non-fullscreen mode this scrolls chrome
# content (status bar, input rules) into terminal scrollback on every render. When the terminal
# column-shrinks, the emulator reflows the previously rendered full-width rows into multiple
# narrower rows that get pushed up — leaving ghost duplicates AND polluting scrollback. Same
# issue as pt #29 (open since 2014), #1675, #1933. Surgical fix: wrap _output_screen_diff so
# that when its internal `if current_height > previous_screen.height` branch fires (the one that
# does the bottom-cursor-move), we make it fall through by inflating previous_screen.height
# first.
import prompt_toolkit.renderer as _pt_renderer
from prompt_toolkit.renderer import _output_screen_diff as _orig_osd
if not getattr(_pt_renderer, "_hermes_osd_patched", False):
_pt_renderer._output_screen_diff = functools.partial(
_hermes_call_output_screen_diff, _orig_osd
)
_pt_renderer._hermes_osd_patched = True
except Exception:
pass
_apply_bracketed_paste_timeout_patch()
self._install_resize_recovery(app)
threading.Thread(target=self._tui_spinner_loop, daemon=True).start()
threading.Thread(target=self._tui_process_loop, daemon=True).start()
# Wake word listener off-thread so a first-run engine install never blocks the prompt.
threading.Thread(target=self._tui_wake_startup, daemon=True, name="wake-startup").start()
atexit.register(_run_cleanup)
self._tui_install_signal_handlers()
if not self._tui_stdin_usable():
_run_cleanup()
self._print_exit_summary()
return
try:
with patch_stdout():
try:
# run_in_terminal() may return either: • a coroutine / Future (prompt_toolkit ≥ 3.0) —
# must be scheduled via ensure_future so the coroutine is actually awaited; calling it
# bare would leave it unawaited and silently drop the output (fixes #23185 Bug A). •
# None (some mocks / older PT builds) — just call the inner function directly since PT
# already executed it synchronously. Do NOT fall back to a bare _pt_print when
# ensure_future raises, because run_in_terminal already invoked the lambda in that case
# (the mock path), which would double-print the line.
import asyncio as _aio
_aio.get_running_loop().set_exception_handler(self._tui_suppress_closed_loop_errors)
except Exception:
pass # no running loop -- nothing to patch
# Record that the app enables focus reporting + mouse tracking so _run_cleanup
# resets them; extended key modes are popped by the same reset.
# When multiline shortcuts are on, also ask supported terminals (e.g. iTerm2) to report
# modified keys distinctly (kitty protocol + modifyOtherKeys); the cleanup reset pops both
# modes. See #36823.
_mark_tui_input_modes_active()
if self._tui_multiline_shortcuts:
_enable_extended_enter_keys(app.output)
self._pet_start_anim()
app.run()
except (EOFError, KeyboardInterrupt, BrokenPipeError):
pass
except (KeyError, OSError) as _stdin_err:
# Selector registration failures from broken stdin and I/O errors from a
# broken stdout during interrupt (EIO is suppressed).
_errno = getattr(_stdin_err, "errno", None) if isinstance(_stdin_err, OSError) else None
_msg = str(_stdin_err)
if _errno == errno.EIO:
pass
elif _errno in {errno.EINVAL, errno.EBADF} or any(
s in _msg for s in ("is not registered", "Bad file descriptor", "Invalid argument")
):
print(
f"\nError: stdin is not usable ({_stdin_err}).\n"
"This can happen with certain Python installations (e.g. uv-managed cPython on macOS)\n"
"where kqueue cannot register fd 0.\n"
"Try reinstalling Python via pyenv or Homebrew, then re-run: hermes setup"
)
else:
raise
finally:
# A resize right before exit leaves its recovery (and the paints it held) unrun.
_release_paints()
self._tui_shutdown()
# /update relaunch happens here, after prompt_toolkit restored terminal modes, on the
# main thread (the process_loop thread would skip cleanup / only exit itself on Windows).
if self._pending_relaunch:
from hermes_cli.relaunch import relaunch
relaunch(self._pending_relaunch, preserve_inherited=False)
def _build_cli_from_args(model, toolsets, provider, reasoning, api_key, base_url, max_turns, run_budget, verbose, compact, resume, checkpoints, pass_session_id, ignore_rules, skills):
"""Resolve the toolset list (explicit / coding posture / platform default), construct HermesCLI, and start the background skills preload."""
toolsets_list = None
if isinstance(toolsets, str) and toolsets:
toolsets_list = [t.strip() for t in toolsets.split(",")]
elif isinstance(toolsets, (list, tuple)) and toolsets:
# Fire may pass multiple --toolsets as a tuple
toolsets_list = []
for t in toolsets:
toolsets_list.extend([x.strip() for x in t.split(",")] if isinstance(t, str) else [str(t)])
elif not toolsets:
# Coding posture inside a code workspace, else the shared platform resolver.
try:
from agent.coding_context import coding_selection
toolsets_list = coding_selection(platform="cli", config=CLI_CONFIG)
except Exception:
toolsets_list = None
if toolsets_list is None:
from hermes_cli.tools_config import _get_platform_tools
toolsets_list = sorted(_get_platform_tools(CLI_CONFIG, "cli"))
parsed_skills = _parse_skills_argument(skills)
try:
cli = HermesCLI(
model=model,
toolsets=toolsets_list,
provider=provider,
reasoning=reasoning,
api_key=api_key,
base_url=base_url,
max_turns=max_turns,
run_budget=run_budget,
verbose=verbose,
compact=compact,
resume=resume,
checkpoints=checkpoints,
pass_session_id=pass_session_id,
ignore_rules=ignore_rules,
)
except ImportError as e:
# Direct `python cli.py` bypasses cmd_chat's partial-update ImportError handler.
from hermes_constants import emit_partial_update_hint
if emit_partial_update_hint(e):
sys.exit(1)
raise
# skills.auto_load rides the same background preload as -s; --ignore-rules skips it with
# the rest of the auto-injected context. Resolved here (not lazily in the agent) so the
# session id is real for ${HERMES_SESSION_ID} and -s can dedupe against it.
from agent.skill_commands import build_auto_load_prompt, resolve_auto_load_skills
auto_load_names = [] if getattr(cli, "ignore_rules", ignore_rules) else resolve_auto_load_skills(CLI_CONFIG)
if not auto_load_names:
cli._auto_load_skills_result = ("", [], [])
if parsed_skills or auto_load_names:
# Load the skill payloads in the background: skill_view walks the full skills
# tree per skill (~0.5s for a large library) and the result is only consumed
# at agent init, not by the banner. finalize_preloaded_skills() joins the
# thread before any consumer reads cli.system_prompt.
def _load_preloaded_skills() -> None:
try:
if auto_load_names:
cli._auto_load_skills_result = build_auto_load_prompt(task_id=cli.session_id, user_config=CLI_CONFIG)
if parsed_skills:
cli._preload_skills_result = build_preloaded_skills_prompt(
parsed_skills, task_id=cli.session_id, excluded_loaded_names=set(cli._auto_load_skills_result[1]))
except Exception as exc: # surfaced by finalize
cli._preload_skills_error = exc
cli._preload_skills_requested = [*auto_load_names, *(s for s in parsed_skills if s not in auto_load_names)]
cli._preload_skills_thread = threading.Thread(target=_load_preloaded_skills, name="skills-preload", daemon=True)
cli._preload_skills_thread.start()
return cli
def _run_legacy_gateway():
"""Legacy `cli.py --gateway` entry: arm the startup watchdog (before importing the gateway graph), then run it."""
import asyncio
with suppress(Exception):
from hermes_startup_watchdog import arm_startup_watchdog
arm_startup_watchdog()
from gateway.run import start_gateway
print("Starting Hermes Gateway (messaging platforms)...")
asyncio.run(start_gateway())
def _start_worktree_setup(list_tools, list_toolsets, worktree, w):
"""Start isolated-worktree creation (+ tool prewarm) in the background.
Returns a join callable that publishes ``_active_worktree``/TERMINAL_CWD and
schedules stale-worktree GC, or None when no worktree is wanted.
"""
if list_tools or list_toolsets or not (worktree or w or CLI_CONFIG.get("worktree", False)):
return None
# Overlap tool discovery with the I/O-bound worktree setup so show_banner() hits a warm
# cache (~0.4s). Only on the -w path: plain `hermes` has no I/O wait to hide.
def _prewarm_tools() -> None:
try:
import model_tools as _mt
_mt.get_tool_definitions(quiet_mode=True)
except Exception:
logger.debug("tool prewarm failed", exc_info=True)
threading.Thread(target=_prewarm_tools, name="tool-prewarm", daemon=True).start()
_sync_base = CLI_CONFIG.get("worktree_sync", True)
_wt_result: dict = {}
def _create_worktree() -> None:
try:
_wt_result["info"] = _setup_worktree(sync_base=_sync_base)
except Exception:
logger.debug("worktree setup failed", exc_info=True)
_wt_result["info"] = None
_wt_thread = threading.Thread(target=_create_worktree, name="worktree-setup", daemon=True)
_wt_thread.start()
def _worktree_maintenance(repo: str) -> None:
_prune_stale_worktrees(repo)
_maintain_pack_health(repo)
def _join_worktree() -> Optional[Dict[str, str]]:
_wt_thread.join(timeout=120)
info = _wt_result.get("info")
if not info:
return info
global _active_worktree
_active_worktree = info
os.environ["TERMINAL_CWD"] = info["path"]
atexit.register(_cleanup_worktree, info)
# GC stale worktrees AFTER _setup_worktree so they never race on git's worktree
# metadata (the new tree is immune: <24h age gate + live pid lock); then repack
# once refs are final so lookups stay fast on multi-agent boxes.
_repo = _git_repo_root()
if _repo:
threading.Thread(target=_worktree_maintenance, args=(_repo,), name="worktree-prune", daemon=True).start()
return info
return _join_worktree
def main(
query: str = None,
q: str = None,
oneshot: bool = False,
image: str = None,
toolsets: str = None,
skills: str | list[str] | tuple[str, ...] = None,
model: str = None,
provider: str = None,
reasoning: str = None,
api_key: str = None,
base_url: str = None,
max_turns: int = None,
run_budget: float = None,
verbose: Optional[bool] = None,
quiet: bool = False,
compact: bool = False,
list_tools: bool = False,
list_toolsets: bool = False,
gateway: bool = False,
resume: str = None,
worktree: bool = False,
w: bool = False,
checkpoints: bool = False,
pass_session_id: bool = False,
output_format: str = "text",
ignore_user_config: bool = False,
ignore_rules: bool = False,
):
"""
Hermes Agent CLI - Interactive AI Assistant
Args:
query: Query to run. On a real TTY this seeds an interactive session
(submitted literally as the first turn); with --oneshot/-Q or a
non-TTY it answers and exits. Alias: -q
q: Shorthand for --query
oneshot: With -q: force the legacy answer-and-exit single-query mode
even on a TTY.
image: Optional local image path to attach to a single query
toolsets: Comma-separated list of toolsets to enable (e.g., "web,terminal")
skills: Comma-separated or repeated list of skills to preload for the session
model: Model to use (default: anthropic/claude-opus-4-20250514)
provider: Inference provider ("auto", "openrouter", "nous", "openai-codex", "zai", "kimi-coding", "minimax", "minimax-cn")
reasoning: Reasoning effort for this run (none|minimal|low|medium|high|xhigh|max|ultra). Overrides agent.reasoning_effort.
api_key: API key for authentication
base_url: Base URL for the API
max_turns: Maximum tool-calling iterations (default: 60)
verbose: Enable verbose logging
compact: Use compact display mode
list_tools: List available tools and exit
list_toolsets: List available toolsets and exit
resume: Resume a previous session by its ID (e.g., 20260225_143052_a1b2c3)
worktree: Run in an isolated git worktree (for parallel agents). Alias: -w
w: Shorthand for --worktree
Examples:
python cli.py # Start interactive mode
python cli.py --toolsets web,terminal # Use specific toolsets
python cli.py --skills hermes-agent-dev,github-auth
python cli.py -q "What is Python?" # Single query mode
python cli.py -q "Describe this" --image ~/storage/shared/Pictures/cat.png
python cli.py --list-tools # List tools and exit
python cli.py --resume 20260225_143052_a1b2c3 # Resume session
python cli.py -w # Start in isolated git worktree
python cli.py -w -q "Fix issue #123" # Single query in worktree
"""
# UTF-8 stdio on Windows before any print (Rich box-drawing would UnicodeEncodeError on cp1252).
with suppress(Exception):
from hermes_cli.stdio import configure_windows_stdio
configure_windows_stdio()
os.environ["HERMES_INTERACTIVE"] = "1" # terminal_tool: interactive sudo prompts with timeout
# The banner names affected plugins; the raw per-name compat warnings would only duplicate it on stderr.
with suppress(Exception):
from hermes_cli.plugin_compat import quiet_for_interactive
quiet_for_interactive()
if gateway:
_run_legacy_gateway()
return
_join_worktree = _start_worktree_setup(list_tools, list_toolsets, worktree, w)
query = query or q
# ``hermes chat`` already validated this; the direct Fire entry point gets the same contract.
if output_format == "stream-json":
if not query:
raise ValueError("--format stream-json requires -q/--query")
quiet = True
cli = _build_cli_from_args(model, toolsets, provider, reasoning, api_key, base_url, max_turns, run_budget,
verbose, compact, resume, checkpoints, pass_session_id, ignore_rules, skills)
# Join the background worktree creation before anything consumes TERMINAL_CWD.
# A requested worktree whose setup failed aborts: never silently run without isolation.
wt_info = _join_worktree() if _join_worktree is not None else None
if _join_worktree is not None and not wt_info:
return
# Inject worktree context into agent's system prompt
if wt_info:
wt_note = (
f"\n\n[System note: You are working in an isolated git worktree at "
f"{wt_info['path']}. Your branch is `{wt_info['branch']}`. "
f"Changes here do not affect the main working tree or other agents. "
f"Remember to commit and push your changes, and create a PR if appropriate. "
f"The original repo is at {wt_info['repo_root']}.]"
)
cli.system_prompt = (cli.system_prompt or "") + wt_note
if list_tools or list_toolsets:
cli.show_banner()
(cli.show_tools if list_tools else cli.show_toolsets)()
sys.exit(0)
atexit.register(_run_cleanup) # interactive mode registers again in run() (idempotent)
_install_single_query_signal_handlers(cli)
if query or image:
_run_single_query_mode(cli, query, image, quiet, oneshot, stream_json=output_format == "stream-json")
return
cli.run()
if __name__ == "__main__":
import fire
fire.Fire(main)
# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.
from prompt_toolkit.layout.menus import CompletionsMenu # noqa: F401,E402
from prompt_toolkit.filters import Condition # noqa: F401,E402
from prompt_toolkit.layout import ConditionalContainer # noqa: F401,E402
from prompt_toolkit.layout.processors import ConditionalProcessor # noqa: F401,E402
from prompt_toolkit.layout.dimension import Dimension # noqa: F401,E402
from prompt_toolkit.history import FileHistory # noqa: F401,E402
from prompt_toolkit.layout import FormattedTextControl # noqa: F401,E402
from prompt_toolkit.layout import HSplit # noqa: F401,E402
from prompt_toolkit.key_binding import KeyBindings # noqa: F401,E402
from prompt_toolkit.layout import Layout # noqa: F401,E402
from prompt_toolkit.styles import Style as PTStyle # noqa: F401,E402
from rich.panel import Panel # noqa: F401,E402
from prompt_toolkit.layout.processors import PasswordProcessor # noqa: F401,E402
from prompt_toolkit.layout.processors import Processor # noqa: F401,E402
from prompt_toolkit.widgets import TextArea # noqa: F401,E402
from prompt_toolkit.layout.processors import Transformation # noqa: F401,E402
from prompt_toolkit.layout import Window # noqa: F401,E402
from prompt_toolkit.layout import WindowAlign # noqa: F401,E402
import base64 # noqa: F401,E402
import concurrent.futures # noqa: F401,E402
import copy # noqa: F401,E402
from rich import box as rich_box # noqa: F401,E402
import tempfile # noqa: F401,E402
def AIAgent(*args, **kwargs):
from run_agent import AIAgent as _AIAgent
return _AIAgent(*args, **kwargs)
def CanonicalUsage(*args, **kwargs):
from agent.usage_pricing import CanonicalUsage as _CanonicalUsage
return _CanonicalUsage(*args, **kwargs)
_PLUGIN_COMPAT_LAZY = {
'DEFAULT_BROWSER_CDP_URL': ('hermes_cli.browser_connect', 'DEFAULT_BROWSER_CDP_URL'),
'HERMES_AGENT_LOGO': ('hermes_cli.banner', 'HERMES_AGENT_LOGO'),
'HERMES_CADUCEUS': ('hermes_cli.banner', 'HERMES_CADUCEUS'),
'SlashCommandAutoSuggest': ('hermes_cli.commands_completion', 'SlashCommandAutoSuggest'),
'SlashCommandCompleter': ('hermes_cli.commands_completion', 'SlashCommandCompleter'),
'build_welcome_banner': ('hermes_cli.banner', 'build_welcome_banner'),
'display_hermes_home': ('hermes_constants', 'display_hermes_home'),
'estimate_usage_cost': ('agent.usage_pricing', 'estimate_usage_cost'),
'get_all_toolsets': ('toolsets', 'get_all_toolsets'),
'get_job': ('cron.jobs', 'get_job'),
'get_toolset_for_tool': ('model_tools', 'get_toolset_for_tool'),
'get_toolset_info': ('toolsets', 'get_toolset_info'),
'init_skin_from_config': ('hermes_cli.skin_engine', 'init_skin_from_config'),
'is_browser_debug_ready': ('hermes_cli.browser_connect', 'is_browser_debug_ready'),
'is_table_divider': ('agent.markdown_tables', 'is_table_divider'),
'looks_like_table_row': ('agent.markdown_tables', 'looks_like_table_row'),
'manual_chrome_debug_command': ('hermes_cli.browser_connect', 'manual_chrome_debug_command'),
'print_config_warnings': ('hermes_cli.config', 'print_config_warnings'),
'prompt_for_secret': ('hermes_cli.callbacks', 'prompt_for_secret'),
'set_friendly_tool_labels': ('agent.display', 'set_friendly_tool_labels'),
'set_tool_preview_max_len': ('agent.display', 'set_tool_preview_max_len'),
'setup_logging': ('hermes_logging', 'setup_logging'),
}
def __getattr__(name): # PEP 562 — lazy so no import cycles
target = _PLUGIN_COMPAT_LAZY.get(name)
if target is None:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
import importlib
from hermes_cli.plugin_compat import warn_once
warn_once(__name__, name, *target)
return getattr(importlib.import_module(target[0]), target[1])
# ---- END PLUGIN-COMPAT ----
Reference in New Issue View Git Blame Copy Permalink