diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index ebf499e8a5..faaa2492d7 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -14,6 +14,7 @@ import sys import threading import time import uuid +from contextlib import suppress from datetime import datetime from urllib.parse import urlparse @@ -24,72 +25,171 @@ from rich.panel import Panel from hermes_constants import display_hermes_home, is_termux as _is_termux_environment from agent.turn_context import extract_api_content_sidecar from hermes_cli.browser_connect import ( - DEFAULT_BROWSER_CDP_URL, - discover_local_cdp_url, - find_free_debug_port, - is_browser_debug_ready, - launch_chrome_debug, - local_port_in_use, - manual_chrome_debug_command, -) + DEFAULT_BROWSER_CDP_URL, discover_local_cdp_url, find_free_debug_port, is_browser_debug_ready, + launch_chrome_debug, local_port_in_use, manual_chrome_debug_command) -def _print_lightpanda_engine_status() -> None: - """``/browser status`` line(s) about ``browser.engine: lightpanda`` — silent unless set; - says whether it is in use or which higher-precedence setting shadows it.""" +# --------------------------------------------------------------------------------------- +# Output helpers. Slash-command text is user-visible: every literal below is load-bearing. +# --------------------------------------------------------------------------------------- + +def _cp(*lines: str) -> None: + """``_cprint`` each line (lazy import: cli.py imports this module).""" + from cli import _cprint + for line in lines: + _cprint(line) + + +def _pr(*lines: str) -> None: + """print() each line.""" + for line in lines: + print(line) + + +def _save(key: str, value) -> bool: + """cli.save_config_value, resolved lazily (cli.py imports this module).""" + from cli import save_config_value + return save_config_value(key, value) + + +def _dim(text: str) -> str: + """Wrap ``text`` in the dim ANSI escape.""" + from cli import _DIM, _RST + return f"{_DIM}{text}{_RST}" + + +def _dim_line(text: str) -> str: + """Two-space indented dim line (the standard slash-command hint shape).""" + return f" {_dim(text)}" + + +def _accent(text: str) -> str: + """Wrap ``text`` in the skin accent ANSI escape.""" + from cli import _ACCENT, _RST + return f"{_ACCENT}{text}{_RST}" + + +def _accent_line(text: str) -> str: + """Two-space indented accent line (the standard slash-command headline shape).""" + return f" {_accent(text)}" + + +def _say_block(*lines: str) -> None: + """print() the lines framed by a blank line above and below (the /browser output style).""" + print() + for line in lines: + print(line) + print() + + +def _command_arg(cmd: str, *, lower: bool = False) -> str: + """Everything after the slash-command word, stripped (optionally lowercased).""" + parts = (cmd or "").strip().split(None, 1) + arg = parts[1].strip() if len(parts) > 1 else "" + return arg.lower() if lower else arg + + +def _shlex_args(cmd: str) -> list: + """Tokens after the command word; falls back to whitespace split on unbalanced quotes.""" + import shlex try: - from tools.browser_tool import lightpanda_engine_status, _using_lightpanda_engine + return shlex.split(cmd)[1:] if cmd else [] + except ValueError: + return (cmd or "").split()[1:] - if not _using_lightpanda_engine(): - return - used, reason = lightpanda_engine_status() - except Exception: - return - if not used: - print(f" ⚠ browser.engine is 'lightpanda' but it is NOT in use: {reason}") - return - print(f" Engine: Lightpanda — {reason} (no screenshots)") - try: - from tools.browser_lightpanda import LIGHTPANDA_INSTALL_HINT, find_lightpanda_binary - lightpanda_bin = find_lightpanda_binary() - except Exception: - return - if lightpanda_bin: - print(f" Binary: {lightpanda_bin}") +def _take_flag(parts: list, flag: str): + """Pop ``flag VALUE`` out of ``parts``: ``(rest, value, ok)``; ok=False when the value is + missing (caller prints usage).""" + if flag not in parts: + return parts, None, True + idx = parts.index(flag) + if idx + 1 >= len(parts): + return parts, None, False + return parts[:idx] + parts[idx + 2:], parts[idx + 1], True + + +def _summarize_paths(paths, limit: int = 5) -> str: + """``a, b, c (+N more)`` for a list of paths.""" + more = f" (+{len(paths) - limit} more)" if len(paths) > limit else "" + return ", ".join(paths[:limit]) + more + + +def _ellipsize(text: str, limit: int) -> str: + """``text[:limit]`` plus ``...`` when truncated.""" + return f"{text[:limit]}{'...' if len(text) > limit else ''}" + + +def _plural(n: int, word: str) -> str: + return f"{n} {word}{'s' if n != 1 else ''}" + + +def _print_capped(lines_text: str, limit: int, tail: str, emit) -> None: + """Emit ``lines_text`` through ``emit``, capped at ``limit`` lines with ``tail`` after.""" + lines = lines_text.splitlines() + if len(lines) > limit: + emit("\n".join(lines[:limit])) + print(f"\n ... ({len(lines) - limit} more lines{tail})") else: - print(f" ⚠ lightpanda binary not found — {LIGHTPANDA_INSTALL_HINT}") + emit(lines_text) -# /cron flag tables: flag -> opts key. Order-sensitive in _parse_flags: bool flags never consume -# a value; --repeat is int-validated separately. +# --------------------------------------------------------------------------------------- +# Small data tables. +# --------------------------------------------------------------------------------------- + +# /cron flag tables: flag -> opts key. Order-sensitive in _parse_cron_flags: bool flags never +# consume a value; --repeat is int-validated separately. _CRON_BOOL_FLAGS = {"--clear-skills": "clear_skills", "--all": "all"} _CRON_LIST_FLAGS = {"--skill": "skills", "--add-skill": "add_skills", "--remove-skill": "remove_skills"} _CRON_VALUE_FLAGS = {"--name": "name", "--deliver": "deliver", "--prompt": "prompt", "--schedule": "schedule"} +# /cron subcommand -> CLICommandsMixin method name. +_CRON_SUBCOMMANDS = { + "list": "_cron_list", "add": "_cron_add", "create": "_cron_add", "edit": "_cron_edit", + **{k: "_cron_job_action" for k in ("pause", "resume", "run", "remove", "rm", "delete")}} _ON_WORDS = {"on", "enable", "true", "1"} _OFF_WORDS = {"off", "disable", "false", "0"} # /busy mode -> what Enter does while Hermes is working (status line / post-set explanation). _BUSY_MODE_SHORT = { - "queue": "queues for next turn", - "steer": "steers into current run (after next tool call)", - "interrupt": "redirects current run immediately", -} + "queue": "queues for next turn", "steer": "steers into current run (after next tool call)", + "interrupt": "redirects current run immediately"} _BUSY_MODE_LONG = { "queue": "Enter will queue follow-up input while Hermes is busy.", "steer": "Enter will steer your message into the current run (after the next tool call).", "interrupt": "Enter will redirect the current run while Hermes is busy; /stop still cancels it.", } - # /fast argument -> (service_tier value, persisted config value) _FAST_TIERS = { - "fast": ("priority", "fast"), "on": ("priority", "fast"), - "normal": (None, "normal"), "off": (None, "normal"), - "auto": ("auto", "auto"), "cold": ("cold", "cold"), + "fast": ("priority", "fast"), "on": ("priority", "fast"), "normal": (None, "normal"), + "off": (None, "normal"), "auto": ("auto", "auto"), "cold": ("cold", "cold")} + +# /reasoning display toggles: arg -> (attr, value, headline, follow-up note) +_REASONING_TOGGLES = { + **dict.fromkeys(("show", "on"), ("show_reasoning", True, "ON", + "Model thinking will be shown during and after each response.")), + **dict.fromkeys(("hide", "off"), ("show_reasoning", False, "OFF", "")), + **dict.fromkeys(("full", "all"), ("reasoning_full", True, "FULL", + "The post-response recap box will print complete thinking.")), + **dict.fromkeys(("clamp", "collapse", "short"), ("reasoning_full", False, "CLAMPED to 10 lines", "")), } +# /bg AIAgent provider-routing kwargs -> HermesCLI attribute carrying the value. +_BG_PROVIDER_KWARGS = { + "providers_allowed": "_providers_only", "providers_ignored": "_providers_ignore", + "providers_order": "_providers_order", "provider_sort": "_provider_sort", + "provider_require_parameters": "_provider_require_params", + "provider_data_collection": "_provider_data_collection", + "openrouter_min_coding_score": "_openrouter_min_coding_score", "fallback_model": "_fallback_model"} + +# /diff argument -> mode (anything else is a path; --stat/stat is the stat flag). +_DIFF_MODES = { + "staged": "staged", "--staged": "staged", "cached": "staged", "--cached": "staged", + "all": "all", "--all": "all", "head": "all", "session": "session"} +_DIFF_LABELS = {"working": "Unstaged", "staged": "Staged", "all": "All (vs HEAD)"} + def _split_scope_flags(raw: str): """``(arg, explicit_global)`` for /reasoning + /fast: session scope by default, ``--global`` @@ -121,19 +221,6 @@ def _toggle_target(arg: str, current: bool): return None -def _command_arg(cmd: str, *, lower: bool = False) -> str: - """Everything after the slash-command word, stripped (optionally lowercased).""" - parts = (cmd or "").strip().split(None, 1) - arg = parts[1].strip() if len(parts) > 1 else "" - return arg.lower() if lower else arg - - -def _summarize_paths(paths, limit: int = 5) -> str: - """``a, b, c (+N more)`` for a list of paths.""" - more = f" (+{len(paths) - limit} more)" if len(paths) > limit else "" - return ", ".join(paths[:limit]) + more - - def _cron_api(**kwargs) -> dict: """Call the cronjob model tool and decode its JSON reply.""" from tools.cronjob_tools import cronjob as cronjob_tool @@ -155,8 +242,7 @@ def _parse_cron_flags(tokens): opts = { "name": None, "deliver": None, "repeat": None, "prompt": None, "schedule": None, "skills": [], "add_skills": [], "remove_skills": [], - "clear_skills": False, "all": False, "positionals": [], - } + "clear_skills": False, "all": False, "positionals": []} i = 0 while i < len(tokens): token = tokens[i] @@ -183,32 +269,19 @@ def _parse_cron_flags(tokens): return opts -# /cron subcommand -> CLICommandsMixin handler (bound late, after the class body). -_CRON_SUBCOMMANDS: dict = {} - - -def _say_block(*lines: str) -> None: - """print() the lines framed by a blank line above and below (the /browser output style).""" - print() - for line in lines: - print(line) - print() - +# --------------------------------------------------------------------------------------- +# Session-switch plumbing shared by /resume and /branch. +# --------------------------------------------------------------------------------------- def _end_current_session(cli, reason: str) -> None: """Flush un-persisted messages, then end the current session row with ``reason``. Best-effort on both steps (the switch proceeds even if the DB write fails).""" if cli.agent: - try: + with suppress(Exception): cli.agent._flush_messages_to_session_db( - cli.conversation_history, conversation_history=cli.conversation_history, - ) - except Exception: - pass - try: + cli.conversation_history, conversation_history=cli.conversation_history) + with suppress(Exception): cli._session_db.end_session(cli.session_id, reason) - except Exception: - pass def _sync_agent_to_session(cli, session_id: str, *, parent_session_id: str, reason: str) -> None: @@ -224,38 +297,34 @@ def _sync_agent_to_session(cli, session_id: str, *, parent_session_id: str, reas if hasattr(cli.agent, "_last_flushed_db_idx"): cli.agent._last_flushed_db_idx = len(cli.conversation_history) if hasattr(cli.agent, "_todo_store"): - try: + with suppress(Exception): from tools.todo_tool import TodoStore cli.agent._todo_store = TodoStore() - except Exception: - pass if hasattr(cli.agent, "_invalidate_system_prompt"): cli.agent._invalidate_system_prompt() - try: + with suppress(Exception): _mm = getattr(cli.agent, "_memory_manager", None) if _mm is not None: _mm.on_session_switch( - session_id, parent_session_id=parent_session_id or "", reset=False, reason=reason, - ) - except Exception: - pass + session_id, parent_session_id=parent_session_id or "", reset=False, reason=reason) + + +def _db_unavailable_line() -> str: + from hermes_state import format_session_db_unavailable + return f" {format_session_db_unavailable()}" def _print_side_result_panel(cli, *, header_lines, body, title_suffix, empty_note) -> None: """Print a worker-thread result (/bg, /btw) into the scrollback: accent rules around ``header_lines``, then ``body`` in a skinned Rich panel (or ``empty_note``). Forces a TUI refresh first so the spinner/status bar don't overlap the output.""" - from cli import ChatConsole, _accent_hex, _cprint, _maybe_remap_for_light_mode, _render_final_assistant_content - if cli._app: - cli._app.invalidate() - time.sleep(0.05) # brief pause for refresh - print() + from cli import ChatConsole, _accent_hex, _maybe_remap_for_light_mode, _render_final_assistant_content + _refresh_tui_before_print(cli) ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]") - for line in header_lines: - _cprint(line) + _cp(*header_lines) ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]") if not body: - _cprint(empty_note) + _cp(empty_note) return try: from hermes_cli.skin_engine import get_active_skin @@ -267,27 +336,58 @@ def _print_side_result_panel(cli, *, header_lines, body, title_suffix, empty_not label, _resp_color, _resp_text = "⚕ Hermes", "#CD7F32", "#FFF8DC" ChatConsole().print(Panel( _render_final_assistant_content(body, mode=cli.final_response_markdown), - title=f"[{_resp_color} bold]{label} {title_suffix}[/]", - title_align="left", - border_style=_resp_color, - style=_resp_text, - box=rich_box.HORIZONTALS, - padding=(1, 4), - width=cli._scrollback_box_width(), - )) + title=f"[{_resp_color} bold]{label} {title_suffix}[/]", title_align="left", + border_style=_resp_color, style=_resp_text, box=rich_box.HORIZONTALS, padding=(1, 4), + width=cli._scrollback_box_width())) + + +def _refresh_tui_before_print(cli) -> None: + """Invalidate the running TUI (brief pause for the redraw) then print a blank separator, so + worker-thread output doesn't overlap the spinner/status bar.""" + if cli._app: + cli._app.invalidate() + time.sleep(0.05) + print() + + +# --------------------------------------------------------------------------------------- +# /browser sub-handlers. +# --------------------------------------------------------------------------------------- + +def _print_lightpanda_engine_status() -> None: + """``/browser status`` line(s) about ``browser.engine: lightpanda`` — silent unless set; + says whether it is in use or which higher-precedence setting shadows it.""" + try: + from tools.browser_tool import lightpanda_engine_status, _using_lightpanda_engine + if not _using_lightpanda_engine(): + return + used, reason = lightpanda_engine_status() + except Exception: + return + if not used: + print(f" ⚠ browser.engine is 'lightpanda' but it is NOT in use: {reason}") + return + print(f" Engine: Lightpanda — {reason} (no screenshots)") + try: + from tools.browser_lightpanda import LIGHTPANDA_INSTALL_HINT, find_lightpanda_binary + lightpanda_bin = find_lightpanda_binary() + except Exception: + return + if lightpanda_bin: + print(f" Binary: {lightpanda_bin}") + else: + print(f" ⚠ lightpanda binary not found — {LIGHTPANDA_INSTALL_HINT}") def _browser_use(cli, arg: str) -> None: """/browser use [off] — toggle Browser Use mode (browser.backend); resets the session.""" from hermes_cli.config import load_config, save_config from tools.registry import invalidate_check_fn_cache - if arg not in {"on", "off"}: _say_block( "Usage: /browser use [off]", " /browser use — switch to Browser Use mode (browser_exec via CLI 3.0)", - " /browser use off — revert to the built-in browser tools", - ) + " /browser use off — revert to the built-in browser tools") return config = load_config() if arg == "on": @@ -302,99 +402,103 @@ def _browser_use(cli, arg: str) -> None: cli.new_session() _say_block(headline, " Session reset. New tool configuration is active.") + +def _normalize_cdp_url(cdp_url: str): + """Validate a /browser connect URL: ``(cdp_url, port)`` or None after printing the error. + A ``/devtools/browser/`` path is kept verbatim; anything else is reduced to the origin.""" + parsed = urlparse(cdp_url if "://" in cdp_url else f"http://{cdp_url}") + if parsed.scheme not in {"http", "https", "ws", "wss"}: + _say_block( + f" ⚠ Unsupported browser url scheme: {parsed.scheme or '(missing)'} " + "(expected one of: http, https, ws, wss)") + return None + try: + port = parsed.port or (443 if parsed.scheme in {"https", "wss"} else 80) + except ValueError: + _say_block(f" ⚠ Invalid port in browser url: {cdp_url}") + return None + if not parsed.hostname: + _say_block(f" ⚠ Missing host in browser url: {cdp_url}") + return None + if parsed.path.startswith("/devtools/browser/"): + return parsed.geturl(), port + return parsed._replace(path="", params="", query="", fragment="").geturl(), port + + +def _launch_default_cdp_browser(port: int): + """Launch a local debug browser for the default CDP URL; returns the discovered CDP url or + None (after printing what went wrong / how to launch manually).""" + import platform as _plat + launch_port = port + if local_port_in_use(port): + launch_port = find_free_debug_port(port) + _pr(f" ⚠ Port {port} is occupied by another application that isn't a CDP browser", + f" (an IDE debugger or dev server may be using it) — launching on port {launch_port} instead...") + else: + print(" Chromium-family browser isn't running with remote debugging — attempting to launch...") + launch = launch_chrome_debug(launch_port, _plat.system()) + if not launch.launched: + print(" ⚠ Could not auto-launch a Chromium-family browser") + if launch.hint: + print(f" {launch.hint}") + chrome_cmd = manual_chrome_debug_command(launch_port, _plat.system()) + if chrome_cmd: + _pr(" Launch a Chromium-family browser manually:", f" {chrome_cmd}") + else: + print(" No supported Chromium-family browser executable found in this environment") + return None + for _wait in range(10): # wait for the DevTools discovery endpoint to come up + found = discover_local_cdp_url(launch_port, timeout=1.0) + if found: + print(f" ✓ Chromium-family browser launched and listening on port {launch_port}") + return found + time.sleep(0.5) + _pr(f" ⚠ Browser launched but port {launch_port} isn't responding yet", + " Try again in a few seconds — the debug instance may still be starting") + return None + + def _browser_connect(cli, cdp_url: str) -> None: """/browser connect [url] — validate the CDP URL, find or launch a debug browser, then point the browser tools at it (BROWSER_CDP_URL) and tell the model.""" - import platform as _plat - - parsed_cdp = urlparse(cdp_url if "://" in cdp_url else f"http://{cdp_url}") - if parsed_cdp.scheme not in {"http", "https", "ws", "wss"}: - _say_block( - f" ⚠ Unsupported browser url scheme: {parsed_cdp.scheme or '(missing)'} " - "(expected one of: http, https, ws, wss)" - ) + normalized = _normalize_cdp_url(cdp_url) + if normalized is None: return - try: - _port = parsed_cdp.port or (443 if parsed_cdp.scheme in {"https", "wss"} else 80) - except ValueError: - _say_block(f" ⚠ Invalid port in browser url: {cdp_url}") - return - if not parsed_cdp.hostname: - _say_block(f" ⚠ Missing host in browser url: {cdp_url}") - return - if parsed_cdp.path.startswith("/devtools/browser/"): - cdp_url = parsed_cdp.geturl() - else: - cdp_url = parsed_cdp._replace(path="", params="", query="", fragment="").geturl() + cdp_url, port = normalized # Clear any existing browser sessions so the next tool call uses the new backend - try: + with suppress(Exception): from tools.browser_tool import cleanup_all_browsers cleanup_all_browsers() - except Exception: - pass - print() # Already serving CDP? For the default-local URL probe both loopbacks: a squatter # on 127.0.0.1: (e.g. an IDE debugger) can push the browser to bind [::1] only. - _is_default = cdp_url == DEFAULT_BROWSER_CDP_URL - if _is_default: - _found = discover_local_cdp_url(_port, timeout=1.0) - _already_open = _found is not None - if _found: - cdp_url = _found + is_default = cdp_url == DEFAULT_BROWSER_CDP_URL + if is_default: + found = discover_local_cdp_url(port, timeout=1.0) + already_open = found is not None + if found: + cdp_url = found else: - _already_open = is_browser_debug_ready(cdp_url, timeout=1.0) - - if _already_open: + already_open = is_browser_debug_ready(cdp_url, timeout=1.0) + if already_open: print(f" ✓ Chromium-family browser is already listening at {cdp_url}") - elif _is_default: - _launch_port = _port - if local_port_in_use(_port): - _launch_port = find_free_debug_port(_port) - print(f" ⚠ Port {_port} is occupied by another application that isn't a CDP browser") - print(f" (an IDE debugger or dev server may be using it) — launching on port {_launch_port} instead...") - else: - print(" Chromium-family browser isn't running with remote debugging — attempting to launch...") - _launch = launch_chrome_debug(_launch_port, _plat.system()) - if _launch.launched: - for _wait in range(10): # wait for the DevTools discovery endpoint to come up - _found = discover_local_cdp_url(_launch_port, timeout=1.0) - if _found: - cdp_url = _found - _already_open = True - break - time.sleep(0.5) - if _already_open: - print(f" ✓ Chromium-family browser launched and listening on port {_launch_port}") - else: - print(f" ⚠ Browser launched but port {_launch_port} isn't responding yet") - print(" Try again in a few seconds — the debug instance may still be starting") - else: - print(" ⚠ Could not auto-launch a Chromium-family browser") - if _launch.hint: - print(f" {_launch.hint}") - chrome_cmd = manual_chrome_debug_command(_launch_port, _plat.system()) - if chrome_cmd: - print(" Launch a Chromium-family browser manually:") - print(f" {chrome_cmd}") - else: - print(" No supported Chromium-family browser executable found in this environment") + elif is_default: + found = _launch_default_cdp_browser(port) + already_open = found is not None + if found: + cdp_url = found else: - print(f" ⚠ Port {_port} is not reachable at {cdp_url}") - - if not _already_open: + print(f" ⚠ Port {port} is not reachable at {cdp_url}") + if not already_open: _say_block("Browser not connected — start a Chromium-family browser with remote debugging and retry /browser connect") return - os.environ["BROWSER_CDP_URL"] = cdp_url # Eagerly start the CDP supervisor so pending_dialogs + frame_tree show up in the next snapshot. - try: + with suppress(Exception): from tools.browser_tool import _ensure_cdp_supervisor # type: ignore[import-not-found] _ensure_cdp_supervisor("default") - except Exception: - pass _say_block("🌐 Browser connected to live Chromium-family browser via CDP", f" Endpoint: {cdp_url}") # Tell the model the CDP browser was made available on purpose. @@ -408,29 +512,34 @@ def _browser_connect(cli, cdp_url: str) -> None: "just because CDP is connected. This is typically a Hermes-managed isolated debug " "profile, not the user's main everyday browser. It is still user-visible and may contain " "pages, logged-in sessions, or cookies in that debug profile, so avoid destructive actions, " - "closing tabs, or navigating away unless the user's task calls for it.]" - ) + "closing tabs, or navigating away unless the user's task calls for it.]") + def _browser_disconnect(cli) -> None: if not os.environ.get("BROWSER_CDP_URL", "").strip(): _say_block("Browser is not connected to a live Chromium-family browser (already using default mode)") return os.environ.pop("BROWSER_CDP_URL", None) - try: + with suppress(Exception): from tools.browser_tool import cleanup_all_browsers, _stop_cdp_supervisor _stop_cdp_supervisor("default") cleanup_all_browsers() - except Exception: - pass - _say_block( - "🌐 Browser disconnected from live Chromium-family browser", - " Browser tools reverted to default mode (local headless or cloud provider)", - ) + _say_block("🌐 Browser disconnected from live Chromium-family browser", + " Browser tools reverted to default mode (local headless or cloud provider)") if hasattr(cli, '_pending_input'): cli._pending_input.put( "[System note: The user has disconnected the browser tools from their live Chromium-family browser. " - "Browser tools are back to default mode (headless local browser or cloud provider).]" - ) + "Browser tools are back to default mode (headless local browser or cloud provider).]") + + +def _browser_tool_attr(name: str, default): + """``tools.browser_tool.()`` or ``default`` when the import/call fails.""" + try: + import tools.browser_tool as bt + return getattr(bt, name)() + except Exception: + return default + def _browser_status() -> None: current = os.environ.get("BROWSER_CDP_URL", "").strip() @@ -441,20 +550,18 @@ def _browser_status() -> None: except Exception: _bu_mode = False if _bu_mode: - print("🌐 Browser: Browser Use mode (browser_exec via the Browser Use CLI 3.0)") - print(" Local Chrome via CDP, or Browser Use cloud browsers") + _pr("🌐 Browser: Browser Use mode (browser_exec via the Browser Use CLI 3.0)", + " Local Chrome via CDP, or Browser Use cloud browsers") _print_lightpanda_engine_status() _say_block(" /browser use off — revert to the built-in browser tools") return if current: - print("🌐 Browser: connected to live Chromium-family browser via CDP") - print(f" Endpoint: {current}") + _pr("🌐 Browser: connected to live Chromium-family browser via CDP", + f" Endpoint: {current}") _print_lightpanda_engine_status() _port = 9222 - try: + with suppress(ValueError, IndexError): _port = int(current.rsplit(":", 1)[-1].split("/")[0]) - except (ValueError, IndexError): - pass try: import socket s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) @@ -462,78 +569,62 @@ def _browser_status() -> None: s.connect(("127.0.0.1", _port)) s.close() print(" Status: ✓ reachable") - except (OSError, Exception): + except Exception: print(" Status: ⚠ not reachable (browser may not be running)") else: - try: - from tools.browser_tool import _get_cloud_provider - provider = _get_cloud_provider() - except Exception: - provider = None + provider = _browser_tool_attr("_get_cloud_provider", None) if provider is not None: print(f"🌐 Browser: {provider.provider_name()} (cloud)") _print_lightpanda_engine_status() else: - try: - from tools.browser_tool import _get_browser_engine - engine = _get_browser_engine() - except Exception: - engine = "auto" + engine = _browser_tool_attr("_get_browser_engine", "auto") if engine == "lightpanda": - print("🌐 Browser: local Lightpanda (agent-browser --engine lightpanda)") - print(" ⚡ Lightpanda: faster navigation, no screenshot support") - print(" Automatic Chromium fallback for screenshots and failed commands") + _pr("🌐 Browser: local Lightpanda (agent-browser --engine lightpanda)", + " ⚡ Lightpanda: faster navigation, no screenshot support", + " Automatic Chromium fallback for screenshots and failed commands") _print_lightpanda_engine_status() elif engine == "chrome": print("🌐 Browser: local headless Chromium (agent-browser --engine chrome)") else: print("🌐 Browser: local headless Chromium (agent-browser)") - _say_block( - " /browser connect — connect to your live Chromium-family browser", - " /browser disconnect — revert to default", - ) + _say_block(" /browser connect — connect to your live Chromium-family browser", + " /browser disconnect — revert to default") class CLICommandsMixin: """Mixin holding the interactive-CLI slash-command handlers.""" + # ---- /rollback ------------------------------------------------------------------------ + + def _checkpoint_manager(self, disabled_lines): + """The agent's checkpoint manager, or None after printing why it is unavailable.""" + if not hasattr(self, 'agent') or not self.agent: + print(" No active agent session.") + return None + mgr = self.agent._checkpoint_mgr + if not mgr.enabled: + for line in disabled_lines: + print(line) + return None + return mgr + def _handle_rollback_command(self, command: str): """Handle /rollback — list, diff, or restore filesystem checkpoints. - Syntax: - /rollback — list checkpoints - /rollback — restore checkpoint N, preserving user hand-edits - (also undoes last chat turn) - /rollback --all — classic full restore (may overwrite your later edits) - /rollback diff — preview changes since checkpoint N - /rollback — restore a single file from checkpoint N""" + ``/rollback`` lists; ```` restores N preserving user hand-edits (and undoes the last + chat turn); `` --all`` is the classic full restore; ``diff `` previews changes since + N; `` `` restores a single file.""" from tools.checkpoint_manager import format_checkpoint_list - - if not hasattr(self, 'agent') or not self.agent: - print(" No active agent session.") + mgr = self._checkpoint_manager(( + " Checkpoints are not enabled.", " Enable with: hermes --checkpoints", + " Or in config.yaml: checkpoints: { enabled: true }")) + if mgr is None: return - - mgr = self.agent._checkpoint_mgr - if not mgr.enabled: - print(" Checkpoints are not enabled.") - print(" Enable with: hermes --checkpoints") - print(" Or in config.yaml: checkpoints: { enabled: true }") - return - cwd = os.getenv("TERMINAL_CWD", os.getcwd()) - parts = command.split() - args = parts[1:] if len(parts) > 1 else [] - + args = command.split()[1:] # --all / --force: classic full restore, overwriting user edits too. - restore_all = False - filtered = [] - for a in args: - if a.lower() in ("--all", "--force"): - restore_all = True - else: - filtered.append(a) - args = filtered - + restore_all = any(a.lower() in ("--all", "--force") for a in args) + args = [a for a in args if a.lower() not in ("--all", "--force")] if not args: # List checkpoints — fall back to the cross-project view when the current # directory has none (writes may have landed under the session cwd, not TERMINAL_CWD). @@ -546,133 +637,102 @@ class CLICommandsMixin: return print(format_checkpoint_list(checkpoints, cwd)) return - - # Handle /rollback diff - if args[0].lower() == "diff": - if len(args) < 2: - print(" Usage: /rollback diff ") - return - checkpoints = mgr.list_checkpoints(cwd) - if not checkpoints: - print(f" No checkpoints found for {cwd}") - return - target_hash = self._resolve_checkpoint_ref(args[1], checkpoints) - if not target_hash: - return - result = mgr.diff(cwd, target_hash) - if result["success"]: - stat = result.get("stat", "") - diff = result.get("diff", "") - if not stat and not diff: - print(" No changes since this checkpoint.") - else: - if stat: - print(f"\n{stat}") - if diff: - # Limit diff output to avoid terminal flood - diff_lines = diff.splitlines() - if len(diff_lines) > 80: - print("\n".join(diff_lines[:80])) - print(f"\n ... ({len(diff_lines) - 80} more lines, showing first 80)") - else: - print(f"\n{diff}") - else: - print(f" ❌ {result['error']}") + is_diff = args[0].lower() == "diff" + if is_diff and len(args) < 2: + print(" Usage: /rollback diff ") return - - # Resolve checkpoint reference (number or hash) checkpoints = mgr.list_checkpoints(cwd) if not checkpoints: print(f" No checkpoints found for {cwd}") return - - target_hash = self._resolve_checkpoint_ref(args[0], checkpoints) + target_hash = self._resolve_checkpoint_ref(args[1 if is_diff else 0], checkpoints) if not target_hash: return - - # Check for file-level restore: /rollback - file_path = args[1] if len(args) > 1 else None - - result = mgr.restore( - cwd, target_hash, file_path=file_path, - safe=not restore_all and not file_path, - ) - if result["success"]: - if file_path: - print(f" ✅ Restored {file_path} from checkpoint {result['restored_to']}: {result['reason']}") - else: - print(f" ✅ Restored to checkpoint {result['restored_to']}: {result['reason']}") - skipped = result.get("skipped_user_edits") or [] - if skipped: - print(f" ↷ Kept your hand-edits: {_summarize_paths(skipped)}") - print(" Use /rollback --all to restore those too.") - oversize = result.get("skipped_oversize") or [] - if oversize: - print(f" ↷ Kept (too large for checkpoints, no stored copy to revert to): {_summarize_paths(oversize)}") - failed = result.get("failed_deletes") or [] - if failed: - print(f" ⚠️ Could not remove (left in place): {_summarize_paths(failed)}") - print(" A pre-rollback snapshot was saved automatically.") - - # Also undo the last conversation turn so the agent's context - # matches the restored filesystem state - if self.conversation_history: - self.undo_last(prefill=False) - print(" Chat turn undone to match restored file state.") + if is_diff: + self._rollback_diff(mgr, cwd, target_hash) else: + file_path = args[1] if len(args) > 1 else None + self._rollback_restore(mgr, cwd, target_hash, file_path, restore_all) + + def _rollback_diff(self, mgr, cwd: str, target_hash: str) -> None: + result = mgr.diff(cwd, target_hash) + if not result["success"]: print(f" ❌ {result['error']}") + return + stat, diff = result.get("stat", ""), result.get("diff", "") + if not stat and not diff: + print(" No changes since this checkpoint.") + return + if stat: + print(f"\n{stat}") + if diff: + # Limit diff output to avoid terminal flood + diff_lines = diff.splitlines() + if len(diff_lines) > 80: + _pr("\n".join(diff_lines[:80]), + f"\n ... ({len(diff_lines) - 80} more lines, showing first 80)") + else: + print(f"\n{diff}") + + def _rollback_restore(self, mgr, cwd, target_hash: str, file_path, restore_all: bool) -> None: + result = mgr.restore(cwd, target_hash, file_path=file_path, + safe=not restore_all and not file_path) + if not result["success"]: + print(f" ❌ {result['error']}") + return + what = f"{file_path} from checkpoint" if file_path else "to checkpoint" + print(f" ✅ Restored {what} {result['restored_to']}: {result['reason']}") + skipped = result.get("skipped_user_edits") or [] + if skipped: + _pr(f" ↷ Kept your hand-edits: {_summarize_paths(skipped)}", + " Use /rollback --all to restore those too.") + oversize = result.get("skipped_oversize") or [] + if oversize: + print(" ↷ Kept (too large for checkpoints, no stored copy to revert to): " + f"{_summarize_paths(oversize)}") + failed = result.get("failed_deletes") or [] + if failed: + print(f" ⚠️ Could not remove (left in place): {_summarize_paths(failed)}") + print(" A pre-rollback snapshot was saved automatically.") + # Also undo the last conversation turn so the agent's context matches the restored files. + if self.conversation_history: + self.undo_last(prefill=False) + print(" Chat turn undone to match restored file state.") + + # ---- /diff ---------------------------------------------------------------------------- def _handle_diff_command(self, command: str): """Handle /diff — show git changes in the working directory. Syntax: ``/diff`` (unstaged + untracked), ``staged``, ``all`` (vs HEAD), ``session`` (everything Hermes changed since the checkpoint baseline), ``[mode] --stat``, ``[mode] ``.""" - import shlex - - try: - parts = shlex.split(command)[1:] # preserves quoted paths - except ValueError: - parts = command.split()[1:] - stat_only = False mode = "working" paths: list[str] = [] - for arg in parts: + for arg in _shlex_args(command): # shlex preserves quoted paths low = arg.lower() if low in ("--stat", "stat"): stat_only = True - elif low in ("staged", "--staged", "cached", "--cached"): - mode = "staged" - elif low in ("all", "--all", "head"): - mode = "all" - elif low == "session": - mode = "session" + elif low in _DIFF_MODES: + mode = _DIFF_MODES[low] else: paths.append(arg) - cwd = os.getenv("TERMINAL_CWD", os.getcwd()) - if mode == "session": self._print_session_diff(cwd, stat_only) return - from tools.working_diff import collect_working_diff - result = collect_working_diff(cwd, mode=mode, paths=paths or None) if not result.get("success"): print(f" {result.get('error', 'Could not generate diff')}") return - - stat = result.get("stat", "") - diff = result.get("diff", "") + stat, diff = result.get("stat", ""), result.get("diff", "") untracked = result.get("untracked", []) if result.get("empty") or (not stat and not diff and not untracked): print(" No changes.") return - - label = {"working": "Unstaged", "staged": "Staged", "all": "All (vs HEAD)"}[mode] if stat: - print(f"\n {label}:") + print(f"\n {_DIFF_LABELS[mode]}:") self._print_diff_text(stat) if untracked and mode in ("working", "all"): print("\n Untracked:") @@ -686,39 +746,26 @@ class CLICommandsMixin: def _print_diff_body(self, diff: str, stat_hint: str, limit: int = 400) -> None: """Print a diff, capped at ``limit`` lines with a pointer to the --stat form.""" - diff_lines = diff.splitlines() print("") - if len(diff_lines) > limit: - self._print_diff_text("\n".join(diff_lines[:limit])) - print(f"\n ... ({len(diff_lines) - limit} more lines — {stat_hint})") - else: - self._print_diff_text(diff) + _print_capped(diff, limit, f" — {stat_hint}", self._print_diff_text) def _print_session_diff(self, cwd: str, stat_only: bool): """Print the cumulative checkpoint-baseline diff (/diff session).""" - if not hasattr(self, 'agent') or not self.agent: - print(" No active agent session.") + mgr = self._checkpoint_manager(( + " Checkpoints are not enabled, so there's no session baseline.", + " Enable with: hermes --checkpoints", + " Or in config.yaml: checkpoints: { enabled: true }", + " (Plain /diff still works — it uses git directly.)")) + if mgr is None: return - - mgr = self.agent._checkpoint_mgr - if not mgr.enabled: - print(" Checkpoints are not enabled, so there's no session baseline.") - print(" Enable with: hermes --checkpoints") - print(" Or in config.yaml: checkpoints: { enabled: true }") - print(" (Plain /diff still works — it uses git directly.)") - return - result = mgr.session_diff(cwd) if not result.get("success"): print(f" {result.get('error', 'Could not generate diff')}") return - - stat = result.get("stat", "") - diff = result.get("diff", "") + stat, diff = result.get("stat", ""), result.get("diff", "") if result.get("empty") or (not stat and not diff): print(" No changes — Hermes hasn't edited any files here yet.") return - if stat: self._print_diff_text(f"\n{stat}") if stat_only or not diff: @@ -738,180 +785,139 @@ class CLICommandsMixin: pass print(text) + # ---- /snapshot ------------------------------------------------------------------------ + def _handle_snapshot_command(self, command: str): """Handle /snapshot — lightweight state snapshots for Hermes config/state. Syntax: ``/snapshot`` (list), ``create [label]``, ``restore ``, ``prune [N]`` (default 20).""" - from hermes_cli.backup import ( - create_quick_snapshot, list_quick_snapshots, - restore_quick_snapshot, prune_quick_snapshots, - ) - from hermes_constants import display_hermes_home - parts = command.split() subcmd = parts[1].lower() if len(parts) > 1 else "list" + handler = { + "list": self._snapshot_list, "ls": self._snapshot_list, "create": self._snapshot_create, + "restore": self._snapshot_restore, "rewind": self._snapshot_restore, + "prune": self._snapshot_prune}.get(subcmd) + if handler is None: + _pr(f" Unknown subcommand: {subcmd}", + " Usage: /snapshot [list|create [label]|restore |prune [N]]") + return + handler(parts) - if subcmd in {"list", "ls"}: + def _snapshot_list(self, parts) -> None: + from hermes_cli.backup import list_quick_snapshots + snaps = list_quick_snapshots() + if not snaps: + _pr(" No state snapshots yet.", " Create one: /snapshot create [label]") + return + print(f" State snapshots ({display_hermes_home()}/state-snapshots/):\n") + _pr(f" {'#':>3} {'ID':<35} {'Files':>5} {'Size':>10} {'Label'}", + f" {'─'*3} {'─'*35} {'─'*5} {'─'*10} {'─'*20}") + for i, s in enumerate(snaps, 1): + size = s.get("total_size", 0) + if size < 1024: + size_str = f"{size} B" + elif size < 1024 * 1024: + size_str = f"{size / 1024:.0f} KB" + else: + size_str = f"{size / 1024 / 1024:.1f} MB" + label = s.get("label") or "" + print(f" {i:3} {s['id']:<35} {s.get('file_count', 0):>5} {size_str:>10} {label}") + + def _snapshot_create(self, parts) -> None: + from hermes_cli.backup import create_quick_snapshot + snap_id = create_quick_snapshot(label=" ".join(parts[2:]) if len(parts) > 2 else None) + print(f" Snapshot created: {snap_id}" if snap_id else " No state files found to snapshot.") + + def _snapshot_restore(self, parts) -> None: + from hermes_cli.backup import list_quick_snapshots, restore_quick_snapshot + if len(parts) < 3: + print(" Usage: /snapshot restore ") + snaps = list_quick_snapshots(limit=1) + if snaps: + print(f" Most recent: {snaps[0]['id']}") + return + snap_id = parts[2] + try: + idx = int(snap_id) # restore by number (1-indexed) + except ValueError: + idx = None + if idx is not None: snaps = list_quick_snapshots() - if not snaps: - print(" No state snapshots yet.") - print(" Create one: /snapshot create [label]") + if not 1 <= idx <= len(snaps): + print(f" Invalid snapshot number. Use 1-{len(snaps)}.") return - print(f" State snapshots ({display_hermes_home()}/state-snapshots/):\n") - print(f" {'#':>3} {'ID':<35} {'Files':>5} {'Size':>10} {'Label'}") - print(f" {'─'*3} {'─'*35} {'─'*5} {'─'*10} {'─'*20}") - for i, s in enumerate(snaps, 1): - size = s.get("total_size", 0) - if size < 1024: - size_str = f"{size} B" - elif size < 1024 * 1024: - size_str = f"{size / 1024:.0f} KB" - else: - size_str = f"{size / 1024 / 1024:.1f} MB" - label = s.get("label") or "" - print(f" {i:3} {s['id']:<35} {s.get('file_count', 0):>5} {size_str:>10} {label}") - - elif subcmd == "create": - label = " ".join(parts[2:]) if len(parts) > 2 else None - snap_id = create_quick_snapshot(label=label) - if snap_id: - print(f" Snapshot created: {snap_id}") - else: - print(" No state files found to snapshot.") - - elif subcmd in {"restore", "rewind"}: - if len(parts) < 3: - print(" Usage: /snapshot restore ") - snaps = list_quick_snapshots(limit=1) - if snaps: - print(f" Most recent: {snaps[0]['id']}") - return - snap_id = parts[2] - # Allow restore by number (1-indexed) - try: - idx = int(snap_id) - snaps = list_quick_snapshots() - if 1 <= idx <= len(snaps): - snap_id = snaps[idx - 1]["id"] - else: - print(f" Invalid snapshot number. Use 1-{len(snaps)}.") - return - except ValueError: - pass - - # Close our SessionDB first so the restore doesn't contend with this process's live connection. - local_session_db = getattr(self, "_session_db", None) - if local_session_db is not None: - try: - local_session_db.close() - self._session_db = None - except Exception: - pass - - if restore_quick_snapshot(snap_id): - print(f" Restored state from: {snap_id}") - print( - " Restart recommended for gateway/dashboard processes " - "to pick up state.db changes." - ) - else: - print(f" Snapshot not found: {snap_id}") - - elif subcmd == "prune": - keep = 20 - if len(parts) > 2: - try: - keep = int(parts[2]) - except ValueError: - print(" Usage: /snapshot prune [keep-count]") - return - deleted = prune_quick_snapshots(keep=keep) - print(f" Pruned {deleted} old snapshot(s) (keeping {keep}).") + snap_id = snaps[idx - 1]["id"] + # Close our SessionDB first so the restore doesn't contend with this process's live connection. + local_session_db = getattr(self, "_session_db", None) + if local_session_db is not None: + with suppress(Exception): + local_session_db.close() + self._session_db = None + if restore_quick_snapshot(snap_id): + _pr(f" Restored state from: {snap_id}", + " Restart recommended for gateway/dashboard processes to pick up state.db changes.") else: - print(f" Unknown subcommand: {subcmd}") - print(" Usage: /snapshot [list|create [label]|restore |prune [N]]") + print(f" Snapshot not found: {snap_id}") + + def _snapshot_prune(self, parts) -> None: + from hermes_cli.backup import prune_quick_snapshots + keep = 20 + if len(parts) > 2: + try: + keep = int(parts[2]) + except ValueError: + print(" Usage: /snapshot prune [keep-count]") + return + deleted = prune_quick_snapshots(keep=keep) + print(f" Pruned {deleted} old snapshot(s) (keeping {keep}).") + + # ---- /export, /import ----------------------------------------------------------------- def _handle_export_command(self, command: str): """Handle /export [profile] [-o path] — export a profile to a shareable .tar.gz archive.""" - from hermes_cli.profiles import ( - export_profile, - get_active_profile_name, - get_profile_export_path, - ) - - parts = command.split()[1:] - output = None - if "-o" in parts: - idx = parts.index("-o") - if idx + 1 >= len(parts): - print(" Usage: /export [profile] [-o output.tar.gz]") - return - output = parts[idx + 1] - parts = parts[:idx] + parts[idx + 2:] - + from hermes_cli.profiles import export_profile, get_active_profile_name, get_profile_export_path + parts, output, ok = _take_flag(command.split()[1:], "-o") + if not ok: + print(" Usage: /export [profile] [-o output.tar.gz]") + return name = parts[0] if parts else (get_active_profile_name() or "default") - try: - if not output: - output = str(get_profile_export_path(name)) - result = export_profile(name, output) - print(f" ✓ Exported '{name}' to {result}") - print(" Share it: the other user runs /import or `hermes profile import `.") + result = export_profile(name, output or str(get_profile_export_path(name))) + _pr(f" ✓ Exported '{name}' to {result}", + " Share it: the other user runs /import or `hermes profile import `.") except (ValueError, FileNotFoundError, OSError) as e: print(f" Error: {e}") def _handle_import_command(self, command: str): - """Handle /import — import a shared profile archive as a new profile. - - Syntax: - /import [--name ] - """ - from hermes_cli.profiles import ( - check_alias_collision, create_wrapper_script, import_profile, - ) - - parts = command.split()[1:] - name = None - if "--name" in parts: - idx = parts.index("--name") - if idx + 1 >= len(parts): - print(" Usage: /import [--name ]") - return - name = parts[idx + 1] - parts = parts[:idx] + parts[idx + 2:] - - if not parts: + """Handle /import [--name ] — import a shared profile archive as a + new profile.""" + from hermes_cli.profiles import check_alias_collision, create_wrapper_script, import_profile + parts, name, ok = _take_flag(command.split()[1:], "--name") + if not ok or not parts: print(" Usage: /import [--name ]") return - - archive = " ".join(parts) # paths may contain spaces - try: - profile_dir = import_profile(archive, name=name) + profile_dir = import_profile(" ".join(parts), name=name) # paths may contain spaces except (ValueError, FileExistsError, FileNotFoundError) as e: print(f" Error: {e}") return - imported = profile_dir.name print(f" ✓ Imported profile '{imported}' at {profile_dir}") - try: + with suppress(Exception): if not check_alias_collision(imported): wrapper_path = create_wrapper_script(imported) if wrapper_path: print(f" Wrapper created: {wrapper_path}") - except Exception: - pass print(f" Use it: hermes -p {imported}") + # ---- /stop, /agents ------------------------------------------------------------------- + def _handle_stop_command(self): """Handle /stop — kill all running background processes and background (async) delegations. Separate from interrupt (stop the current turn), as in Codex.""" from tools.process_registry import process_registry - - processes = process_registry.list_sessions() - running = [p for p in processes if p.get("status") == "running"] - + running = [p for p in process_registry.list_sessions() if p.get("status") == "running"] # Background subagents live in their own registry, not the process registry. try: from tools.async_delegation import active_count, interrupt_all @@ -919,36 +925,27 @@ class CLICommandsMixin: except Exception: n_async = 0 interrupt_all = None - if not running and not n_async: print(" No running background processes.") return - if running: print(f" Stopping {len(running)} background process(es)...") - killed = process_registry.kill_all() - print(f" ✅ Stopped {killed} process(es).") + print(f" ✅ Stopped {process_registry.kill_all()} process(es).") if n_async and interrupt_all is not None: - stopped = interrupt_all(reason="/stop") - print(f" ✅ Interrupted {stopped} background delegation(s).") + print(f" ✅ Interrupted {interrupt_all(reason='/stop')} background delegation(s).") def _handle_agents_command(self): """Handle /agents — show background processes and agent status.""" - from cli import _cprint from tools.process_registry import format_uptime_short, process_registry - processes = process_registry.list_sessions() running = [p for p in processes if p.get("status") == "running"] finished = [p for p in processes if p.get("status") != "running"] - - _cprint(f" Running processes: {len(running)}") + _cp(f" Running processes: {len(running)}") for p in running: - cmd = p.get("command", "")[:80] up = format_uptime_short(p.get("uptime_seconds", 0)) - _cprint(f" {p.get('session_id', '?')} · {up} · {cmd}") - + _cp(f" {p.get('session_id', '?')} · {up} · {p.get('command', '')[:80]}") if finished: - _cprint(f" Recently finished: {len(finished)}") + _cp(f" Recently finished: {len(finished)}") # Background (async) delegations — delegate_task(background=true) try: @@ -956,47 +953,36 @@ class CLICommandsMixin: delegations = list_async_delegations() except Exception: delegations = [] - running_d = [ - d for d in delegations - if d.get("status") in ("running", "stalling") - ] if delegations: - _cprint(f" Background delegations: {len(running_d)} running") + running_d = [d for d in delegations if d.get("status") in ("running", "stalling")] + _cp(f" Background delegations: {len(running_d)} running") for d in delegations: - goal = (d.get("goal") or "")[:60] status = d.get("status", "?") - line = ( - f" {d.get('delegation_id', '?')} · " - f"{status} · {goal}" - ) - # Live-status detail for in-flight delegations (#51690). + line = f" {d.get('delegation_id', '?')} · {status} · {(d.get('goal') or '')[:60]}" + # Live-status detail for in-flight delegations. if status == "stalling": quiet = d.get("stalled_after_quiet_seconds") if quiet is not None: - line += ( - f" · no progress {quiet:.0f}s — interrupting" - ) - elif status in ("running",): + line += f" · no progress {quiet:.0f}s — interrupting" + elif status == "running": quiet = d.get("seconds_since_progress") if quiet is not None and quiet >= 60: line += f" · quiet {quiet:.0f}s" - _cprint(line) + _cp(line) for i, child in enumerate(d.get("children_activity") or []): if not isinstance(child, dict): continue tool = child.get("current_tool") doing = f"in {tool}" if tool else "between turns" - part = ( - f" └ child {i + 1}: " - f"{child.get('api_calls', '?')} api calls · {doing}" - ) + part = f" └ child {i + 1}: {child.get('api_calls', '?')} api calls · {doing}" idle = child.get("seconds_since_activity") if idle is not None: part += f" · last activity {idle:.0f}s ago" - _cprint(part) - + _cp(part) agent_running = getattr(self, "_agent_running", False) - _cprint(f" Agent: {'running' if agent_running else 'idle'}") + _cp(f" Agent: {'running' if agent_running else 'idle'}") + + # ---- /journey, /paste, /copy, /image -------------------------------------------------- def _handle_journey_command(self, cmd_original: str) -> None: """Handle /journey — the learning timeline (see `hermes journey`). Read-only views render @@ -1006,152 +992,122 @@ class CLICommandsMixin: import io import shlex from contextlib import redirect_stdout - - from cli import _cprint from hermes_cli.journey import register_cli - parser = argparse.ArgumentParser(prog="/journey", add_help=False) register_cli(parser) - rest = cmd_original.split(None, 1) try: - args = parser.parse_args(shlex.split(rest[1]) if len(rest) > 1 else []) + args = parser.parse_args(shlex.split(cmd_original)[1:]) except SystemExit: return - - interactive = getattr(args, "journey_action", None) in ("delete", "edit") try: - if interactive: + if getattr(args, "journey_action", None) in ("delete", "edit"): args.func(args) return args.force_color = True buf = io.StringIO() with redirect_stdout(buf): args.func(args) - _cprint(buf.getvalue().rstrip("\n")) + _cp(buf.getvalue().rstrip("\n")) except Exception as exc: - _cprint(f" /journey failed: {exc}") + _cp(f" /journey failed: {exc}") def _handle_paste_command(self): """Handle /paste — explicitly check the clipboard for an image; the reliable fallback where BracketedPaste doesn't fire for image-only clipboards (VSCode terminal, Windows Terminal/WSL2).""" - from cli import _DIM, _RST, _cprint, _termux_example_image_path + from cli import _termux_example_image_path if _is_termux_environment(): - _cprint( - f" {_DIM}Clipboard image paste is not available on Termux — " - f"use /image or paste a local image path like " - f"{_termux_example_image_path()}{_RST}" - ) + _cp(_dim_line("Clipboard image paste is not available on Termux — use /image or " + f"paste a local image path like {_termux_example_image_path()}")) return - from hermes_cli.clipboard import has_clipboard_image - if has_clipboard_image(): - if self._try_attach_clipboard_image(): - n = len(self._attached_images) - _cprint(f" 📎 Image #{n} attached from clipboard") - else: - _cprint(f" {_DIM}(>_<) Clipboard has an image but extraction failed{_RST}") + if not has_clipboard_image(): + _cp(_dim_line('(._.) No image found in clipboard')) + elif self._try_attach_clipboard_image(): + _cp(f" 📎 Image #{len(self._attached_images)} attached from clipboard") else: - _cprint(f" {_DIM}(._.) No image found in clipboard{_RST}") + _cp(_dim_line('(>_<) Clipboard has an image but extraction failed')) def _handle_copy_command(self, cmd_original: str) -> None: """Handle /copy [number] — copy assistant output to clipboard.""" - from cli import _assistant_copy_text, _cprint - parts = cmd_original.split(maxsplit=1) - arg = parts[1].strip() if len(parts) > 1 else "" - + from cli import _assistant_copy_text + arg = _command_arg(cmd_original) assistant = [m for m in self.conversation_history if m.get("role") == "assistant"] if not assistant: - _cprint(" Nothing to copy yet.") + _cp(" Nothing to copy yet.") return - if arg: try: idx = int(arg) - 1 except ValueError: - _cprint(" Usage: /copy [number]") + _cp(" Usage: /copy [number]") return if idx < 0 or idx >= len(assistant): - _cprint(f" Invalid response number. Use 1-{len(assistant)}.") + _cp(f" Invalid response number. Use 1-{len(assistant)}.") return else: idx = len(assistant) - 1 while idx >= 0 and not _assistant_copy_text(assistant[idx].get("content")): idx -= 1 if idx < 0: - _cprint(" Nothing to copy in assistant responses yet.") + _cp(" Nothing to copy in assistant responses yet.") return - text = _assistant_copy_text(assistant[idx].get("content")) if not text: - _cprint(" Nothing to copy in that assistant response.") + _cp(" Nothing to copy in that assistant response.") return - try: - from hermes_cli.clipboard import ( - is_remote_shell_session, - write_clipboard_text, - ) - if is_remote_shell_session(): - # Over SSH native tools write the REMOTE clipboard; OSC 52 reaches the user's terminal. + from hermes_cli.clipboard import is_remote_shell_session, write_clipboard_text + # Over SSH native tools write the REMOTE clipboard; OSC 52 reaches the user's terminal. + # Locally, OSC 52 is the fallback when native tools are unavailable/fail (SSH/tmux). + if is_remote_shell_session() or not write_clipboard_text(text): self._write_osc52_clipboard(text) - _cprint( - f" Copied assistant response #{idx + 1} via OSC 52 " - "(terminal support required)" - ) - return - if write_clipboard_text(text): - _cprint(f" Copied assistant response #{idx + 1} to clipboard") - return - # Native tools unavailable/failed — OSC 52 fallback (SSH/tmux via the emulator). - self._write_osc52_clipboard(text) - _cprint( - f" Copied assistant response #{idx + 1} via OSC 52 " - "(terminal support required)" - ) + _cp(f" Copied assistant response #{idx + 1} via OSC 52 (terminal support required)") + else: + _cp(f" Copied assistant response #{idx + 1} to clipboard") except Exception as e: - _cprint(f" Clipboard copy failed: {e}") + _cp(f" Clipboard copy failed: {e}") def _handle_image_command(self, cmd_original: str): """Handle /image — attach a local image file for the next prompt.""" - from cli import _DIM, _IMAGE_EXTENSIONS, _RST, _cprint, _resolve_attachment_path, _split_path_input, _termux_example_image_path + from cli import ( + _IMAGE_EXTENSIONS, _resolve_attachment_path, _split_path_input, _termux_example_image_path) raw_args = (cmd_original.split(None, 1)[1].strip() if " " in cmd_original else "") if not raw_args: hint = _termux_example_image_path() if _is_termux_environment() else "/path/to/image.png" - _cprint(f" {_DIM}Usage: /image e.g. /image {hint}{_RST}") + _cp(_dim_line(f'Usage: /image e.g. /image {hint}')) return - path_token, _remainder = _split_path_input(raw_args) image_path = _resolve_attachment_path(path_token) if image_path is None: - _cprint(f" {_DIM}(>_<) File not found: {path_token}{_RST}") + _cp(_dim_line(f'(>_<) File not found: {path_token}')) return if image_path.suffix.lower() not in _IMAGE_EXTENSIONS: - _cprint(f" {_DIM}(._.) Not a supported image file: {image_path.name}{_RST}") + _cp(_dim_line(f'(._.) Not a supported image file: {image_path.name}')) return - self._attached_images.append(image_path) - _cprint(f" 📎 Attached image: {image_path.name}") + _cp(f" 📎 Attached image: {image_path.name}") if _remainder: - _cprint(f" {_DIM}Now type your prompt (or use --image in single-query mode): {_remainder}{_RST}") + _cp(_dim_line(f'Now type your prompt (or use --image in single-query mode): {_remainder}')) elif _is_termux_environment(): - _cprint(f" {_DIM}Tip: type your next message, or run hermes chat -q --image {_termux_example_image_path(image_path.name)} \"What do you see?\"{_RST}") + example = _termux_example_image_path(image_path.name) + tip = f'Tip: type your next message, or run hermes chat -q --image {example} "What do you see?"' + _cp(_dim_line(tip)) + + # ---- /tools, /profile ----------------------------------------------------------------- def _handle_tools_command(self, cmd: str): """Handle /tools [list|disable|enable]. Bare shows the tool list; ``list`` shows per-toolset status; disable/enable save to config and reset the session so the new tool set takes effect cleanly (no prompt-cache breakage mid-conversation).""" - from cli import _ACCENT, _DIM, _RST, _cprint - import shlex from argparse import Namespace from contextlib import redirect_stdout from io import StringIO from hermes_cli.tools_config import tools_disable_enable_command def _run_capture(ns: Namespace) -> None: - """Run tools_disable_enable_command, routing its ANSI print() output through _cprint - inside the interactive TUI so patch_stdout's StdoutProxy doesn't garble the escapes. - Standalone/tests call straight through so real stdout / pytest capture works.""" - # Standalone/tests, run as usual + """Inside the interactive TUI, route the ANSI print() output through _cprint so + patch_stdout's StdoutProxy doesn't garble the escapes; standalone/tests call straight + through so real stdout / pytest capture works.""" if getattr(self, "_app", None) is None: tools_disable_enable_command(ns) return @@ -1160,59 +1116,47 @@ class CLICommandsMixin: class _TTYBuf(StringIO): def isatty(self) -> bool: return True - buf = _TTYBuf() with redirect_stdout(buf): tools_disable_enable_command(ns) for line in buf.getvalue().splitlines(): - _cprint(line) - - try: - parts = shlex.split(cmd) - except ValueError: - parts = cmd.split() - - subcommand = parts[1] if len(parts) > 1 else "" + _cp(line) + parts = _shlex_args(cmd) + subcommand = parts[0] if parts else "" if subcommand not in {"list", "disable", "enable"}: self.show_tools() return - if subcommand == "list": _run_capture(Namespace(tools_action="list", platform="cli")) return - - names = parts[2:] + names = parts[1:] if not names: - print(f"(._.) Usage: /tools {subcommand} [name ...]") - print(f" Built-in toolset: /tools {subcommand} web") - print(f" MCP tool: /tools {subcommand} github:create_issue") + _pr(f"(._.) Usage: /tools {subcommand} [name ...]", + f" Built-in toolset: /tools {subcommand} web", + f" MCP tool: /tools {subcommand} github:create_issue") return # Typing the command is consent. Do NOT use input() — it hangs in prompt_toolkit's loop. verb = "Disabling" if subcommand == "disable" else "Enabling" - label = ", ".join(names) - _cprint(f"{_ACCENT}{verb} {label}...{_RST}") - + _cp(_accent(f"{verb} {', '.join(names)}...")) _run_capture(Namespace(tools_action=subcommand, names=names, platform="cli")) - from hermes_cli.tools_config import _get_platform_tools from hermes_cli.config import load_config self.enabled_toolsets = _get_platform_tools(load_config(), "cli") self.new_session() - _cprint(f"{_DIM}Session reset. New tool configuration is active.{_RST}") + _cp(_dim("Session reset. New tool configuration is active.")) def _handle_profile_command(self): """Display active profile name and home directory.""" from hermes_cli.slash_exec import CommandContext, execute_command - reply = execute_command("profile", CommandContext(surface="cli")) - profile_name = reply.data["profile"] - display = reply.data["home"] + _say_block(f" Profile: {reply.data['profile']}", f" Home: {reply.data['home']}") - print() - print(f" Profile: {profile_name}") - print(f" Home: {display}") - print() + # ---- /handoff ------------------------------------------------------------------------- + + _HANDOFF_PENDING_TIMEOUT = 60.0 + _HANDOFF_RUNNING_TIMEOUT = 900.0 # full synthetic agent turn + delivery + _HANDOFF_HEARTBEAT_EVERY = 30.0 def _handle_handoff_command(self, cmd_original: str) -> bool: """Handle ``/handoff `` — transfer this CLI session to a gateway platform. @@ -1223,37 +1167,43 @@ class CLICommandsMixin: dispatch to finish. ``completed`` → resume hint + return False (caller exits like /quit). ``failed`` / pending-timeout → error + True (session kept). A running-timeout leaves the row untouched (the gateway owns it) and returns True.""" - from cli import _cprint - from hermes_state import format_session_db_unavailable - - parts = cmd_original.split(maxsplit=1) - if len(parts) < 2 or not parts[1].strip(): - _cprint(" Usage: /handoff ") - _cprint(" Hands the current session off to that platform's home channel.") - _cprint(" The CLI session ends here; resume it later with /resume.") + platform_name = _command_arg(cmd_original).lower() + if not platform_name: + _cp(" Usage: /handoff ", + " Hands the current session off to that platform's home channel.", + " The CLI session ends here; resume it later with /resume.") return True + home = self._handoff_validate_target(platform_name) + if home is None: + return True + session_title = self._handoff_prepare_session() + if session_title is None: + return True + if not self._session_db.request_handoff(self.session_id, platform_name): + _cp(" Session is already in flight for handoff. Wait for it to settle, then retry.") + return True + _cp(f" Queued handoff of '{session_title}' → {platform_name} (home: {home.name}).", + " Waiting for the gateway to pick it up...") + return self._handoff_wait(platform_name, session_title) - platform_name = parts[1].strip().lower() - - # Validate platform name + home channel via the live gateway config. + def _handoff_validate_target(self, platform_name: str): + """Resolve the destination home channel via the live gateway config; None (after printing + the reason) when the platform is unknown, disabled, or has no home channel.""" try: from gateway.config import load_gateway_config, Platform except Exception as exc: # pragma: no cover — gateway pkg always shipped - _cprint(f" Could not load gateway config: {exc}") - return True - + _cp(f" Could not load gateway config: {exc}") + return None try: platform = Platform(platform_name) except (ValueError, KeyError): - _cprint(f" Unknown platform '{platform_name}'.") - return True - + _cp(f" Unknown platform '{platform_name}'.") + return None try: gw_config = load_gateway_config() except Exception as exc: - _cprint(f" Could not load gateway config: {exc}") - return True - + _cp(f" Could not load gateway config: {exc}") + return None pcfg = gw_config.platforms.get(platform) if not pcfg or not pcfg.enabled: # Relay aliasing: a relay-fronted gateway has only a RELAY config block, yet @@ -1264,80 +1214,57 @@ class CLICommandsMixin: from gateway.relay import relay_platform_identities relay_cfg = gw_config.platforms.get(Platform.RELAY) if relay_cfg and relay_cfg.enabled: - fronted = {p for p, _ in relay_platform_identities()} - relay_fronts = platform_name in fronted + relay_fronts = platform_name in {p for p, _ in relay_platform_identities()} except Exception: relay_fronts = False if not relay_fronts: - _cprint(f" Platform '{platform_name}' is not configured/enabled in the gateway.") - return True - + _cp(f" Platform '{platform_name}' is not configured/enabled in the gateway.") + return None home = gw_config.get_home_channel(platform) if not home or not home.chat_id: - _cprint(f" No home channel configured for {platform_name}.") - _cprint(" Set one with /sethome on the destination chat first.") - return True + _cp(f" No home channel configured for {platform_name}.", + " Set one with /sethome on the destination chat first.") + return None + return home - # Refuse mid-turn: an in-flight agent run would race with the - # gateway's switch_session and the synthetic turn dispatch. + def _handoff_prepare_session(self): + """Refuse mid-turn, make sure a SessionDB handle + session row exist, and return the + display title (None after printing why the handoff cannot start).""" + # An in-flight agent run would race the gateway's switch_session and the synthetic turn. if getattr(self, "_agent_running", False): - _cprint(" Agent is busy. Wait for the current turn to finish, then retry /handoff.") - return True - - # Make sure we have a SessionDB handle. + _cp(" Agent is busy. Wait for the current turn to finish, then retry /handoff.") + return None if not self._session_db: - try: + with suppress(Exception): from hermes_state import SessionDB self._session_db = SessionDB() - except Exception: - pass if not self._session_db: - _cprint(f" {format_session_db_unavailable()}") - return True - - # Ensure the session row exists (an empty session has flushed nothing yet): the - # gateway needs a row to switch_session onto; set_session_title's INSERT OR IGNORE - # creates it. + _cp(_db_unavailable_line()) + return None + # Ensure the session row exists (an empty session has flushed nothing yet): the gateway + # needs a row to switch_session onto; set_session_title's INSERT OR IGNORE creates it. try: - row = self._session_db.get_session(self.session_id) - if not row: - placeholder_title = f"handoff-{self.session_id[:8]}" - self._session_db.set_session_title(self.session_id, placeholder_title) + if not self._session_db.get_session(self.session_id): + self._session_db.set_session_title(self.session_id, f"handoff-{self.session_id[:8]}") except Exception as exc: - _cprint(f" Could not ensure session row in state.db: {exc}") - return True - - # Display title for messaging. + _cp(f" Could not ensure session row in state.db: {exc}") + return None session_title = "" - try: + with suppress(Exception): row = self._session_db.get_session(self.session_id) if row: session_title = row.get("title") or "" - except Exception: - pass - if not session_title: - session_title = self.session_id[:8] + return session_title or self.session_id[:8] - # Mark pending — gateway watcher will pick this up. - ok = self._session_db.request_handoff(self.session_id, platform_name) - if not ok: - _cprint(" Session is already in flight for handoff. Wait for it to settle, then retry.") - return True - - _cprint(f" Queued handoff of '{session_title}' → {platform_name} (home: {home.name}).") - _cprint(" Waiting for the gateway to pick it up...") - - # Two-phase poll, tick every 0.5s. PENDING (unclaimed): 60s — a timeout means no - # gateway watcher is looking at this state.db, and the CAS fail can't stomp a claim - # landing the same instant. RUNNING (claimed): the gateway replays the full - # transcript via a synthetic turn (routinely >60s), so wait much longer with a - # heartbeat and on timeout do NOT touch the row — failing it here was the - # split-brain bug (CLI said "failed" while the gateway completed the switch). + def _handoff_wait(self, platform_name: str, session_title: str) -> bool: + """Two-phase poll, tick every 0.5s. PENDING (unclaimed): 60s — a timeout means no gateway + watcher is looking at this state.db, and the CAS fail can't stomp a claim landing the same + instant. RUNNING (claimed): the gateway replays the full transcript via a synthetic turn + (routinely >60s), so wait much longer with a heartbeat and on timeout do NOT touch the row + — failing it here was the split-brain bug (CLI said "failed" while the gateway completed + the switch).""" import time as _time - _PENDING_TIMEOUT = 60.0 - _RUNNING_TIMEOUT = 900.0 # full synthetic agent turn + delivery - _HEARTBEAT_EVERY = 30.0 - pending_deadline = _time.time() + _PENDING_TIMEOUT + pending_deadline = _time.time() + self._HANDOFF_PENDING_TIMEOUT running_deadline = None next_heartbeat = None last_state = "pending" @@ -1349,27 +1276,24 @@ class CLICommandsMixin: current = (state_row or {}).get("state") or "pending" if current != last_state: if current == "running": - _cprint(" Gateway picked it up; transferring...") - running_deadline = _time.time() + _RUNNING_TIMEOUT - next_heartbeat = _time.time() + _HEARTBEAT_EVERY + _cp(" Gateway picked it up; transferring...") + running_deadline = _time.time() + self._HANDOFF_RUNNING_TIMEOUT + next_heartbeat = _time.time() + self._HANDOFF_HEARTBEAT_EVERY last_state = current if current == "completed": - _cprint("") - _cprint(f" ↻ Handoff complete. The session is now active on {platform_name}.") - _cprint(f" Resume it on this CLI later with: /resume {session_title}") - _cprint("") - # Mark handed off so _run_cleanup does NOT finalize it on exit: the gateway - # owns the row now, and setting end_reason under it would drop the handoff - # leg from session history / session_search. + _cp("", f" ↻ Handoff complete. The session is now active on {platform_name}.", + f" Resume it on this CLI later with: /resume {session_title}", "") + # Mark handed off so _run_cleanup does NOT finalize it on exit: the gateway owns + # the row now, and setting end_reason under it would drop the handoff leg from + # session history / session_search. from cli import _handed_off_session_ids _handed_off_session_ids.add(self.session_id) - # End the CLI cleanly — same exit semantics as /quit. - self._should_exit = True + self._should_exit = True # same exit semantics as /quit return False if current == "failed": err = (state_row or {}).get("error") or "unknown error" - _cprint(f" Handoff failed: {err}") - _cprint(" Your CLI session is intact. Try /handoff again, or /resume on the platform manually.") + _cp(f" Handoff failed: {err}", + " Your CLI session is intact. Try /handoff again, or /resume on the platform manually.") return True now = _time.time() if current == "pending": @@ -1377,14 +1301,14 @@ class CLICommandsMixin: break else: # running if next_heartbeat is not None and now >= next_heartbeat: - _cprint(" Still transferring (the agent is replaying your session on the destination)...") - next_heartbeat = now + _HEARTBEAT_EVERY + _cp(" Still transferring (the agent is replaying your session on the destination)...") + next_heartbeat = now + self._HANDOFF_HEARTBEAT_EVERY if running_deadline is not None and now >= running_deadline: # Do NOT fail the row: the gateway owns it (split-brain bug otherwise). - _cprint(" The gateway is taking unusually long to finish the transfer.") - _cprint(f" Check {platform_name} — the session may still arrive there.") - _cprint(" This CLI is no longer waiting. Avoid continuing this session here;") - _cprint(" if nothing arrives, retry /handoff once the state settles.") + _cp(" The gateway is taking unusually long to finish the transfer.", + f" Check {platform_name} — the session may still arrive there.", + " This CLI is no longer waiting. Avoid continuing this session here;", + " if nothing arrives, retry /handoff once the state settles.") return True _time.sleep(0.5) @@ -1392,50 +1316,42 @@ class CLICommandsMixin: # instant wins; we lose only the retry convenience, never the handoff). try: self._session_db.fail_handoff( - self.session_id, - "timed out waiting for gateway", - only_states=("pending",), - ) + self.session_id, "timed out waiting for gateway", only_states=("pending",)) except TypeError: # Older SessionDB without only_states (mixed installs): legacy unconditional fail. - try: + with suppress(Exception): self._session_db.fail_handoff(self.session_id, "timed out waiting for gateway") - except Exception: - pass except Exception: pass - _cprint(" Timed out waiting for the gateway. Is `hermes gateway` running?") - _cprint(" Your CLI session is intact.") + _cp(" Timed out waiting for the gateway. Is `hermes gateway` running?", + " Your CLI session is intact.") return True + # ---- /resume, /sessions, /branch ------------------------------------------------------ + def _handle_resume_command(self, cmd_original: str) -> None: """Handle /resume — switch to a previous session mid-conversation.""" - from cli import _cprint, _sync_process_session_id - parts = cmd_original.split(None, 1) - target = parts[1].strip() if len(parts) > 1 else "" + from cli import _sync_process_session_id + target = _command_arg(cmd_original) # Users copy the help text's placeholder brackets/quotes verbatim (``/resume ``). if len(target) >= 2 and (target[0], target[-1]) in {("<", ">"), ("[", "]"), ('"', '"'), ("'", "'")}: target = target[1:-1].strip() - if not target: - _cprint(" Usage: /resume ") + _cp(" Usage: /resume ") if self._show_recent_sessions(reason="resume"): # Arm a one-shot pending-resume selection so a bare number on the next line # works. Must be the same list _show_recent_sessions showed and the numbered # branch below resolves — all three use _list_recent_sessions(limit=10). self._pending_resume_sessions = self._list_recent_sessions(limit=10) return - _cprint(" Tip: Use /history or `hermes sessions list` to find sessions.") + _cp(" Tip: Use /history or `hermes sessions list` to find sessions.") return - # Any explicit /resume supersedes a previously-armed bare - # numbered prompt. + # Any explicit /resume supersedes a previously-armed bare numbered prompt. self._pending_resume_sessions = None - if not self._session_db: - from hermes_state import format_session_db_unavailable - _cprint(f" {format_session_db_unavailable()}") + _cp(_db_unavailable_line()) return # Resolve numbered selection, title, or ID @@ -1443,87 +1359,67 @@ class CLICommandsMixin: sessions = self._list_recent_sessions(limit=10) index = int(target) if index < 1 or index > len(sessions): - _cprint(f" Resume index {index} is out of range.") - _cprint(" Use /resume with no arguments to see available sessions.") + _cp(f" Resume index {index} is out of range.", + " Use /resume with no arguments to see available sessions.") return - selected = sessions[index - 1] - target_id = selected["id"] + target_id = sessions[index - 1]["id"] else: from hermes_cli.main import _resolve_session_by_name_or_id - resolved = _resolve_session_by_name_or_id(target) - target_id = resolved or target - + target_id = _resolve_session_by_name_or_id(target) or target session_meta = self._session_db.get_session(target_id) if not session_meta: - _cprint(f" Session not found: {target}") - _cprint(" Use /sessions or `hermes sessions list` to see available sessions.") + _cp(f" Session not found: {target}", + " Use /sessions or `hermes sessions list` to see available sessions.") return - # If the target is the empty head of a compression chain, redirect to - # the descendant that actually holds the transcript. See #15000. + # If the target is the empty head of a compression chain, redirect to the descendant + # that actually holds the transcript. try: resolved_id = self._session_db.resolve_resume_session_id(target_id) except Exception: resolved_id = target_id if resolved_id and resolved_id != target_id: - _cprint( - f" Session {target_id} was compressed into {resolved_id}; " - f"resuming the descendant with your transcript." - ) + _cp(f" Session {target_id} was compressed into {resolved_id}; " + f"resuming the descendant with your transcript.") target_id = resolved_id - resolved_meta = self._session_db.get_session(target_id) - if resolved_meta: - session_meta = resolved_meta - + session_meta = self._session_db.get_session(target_id) or session_meta if target_id == self.session_id: - _cprint(" Already on that session.") + _cp(" Already on that session.") return - old_session_id = self.session_id _end_current_session(self, "resumed_other") - - # Switch to the target session self.session_id = target_id self._resumed = True self._pending_title = None _sync_process_session_id(target_id) - # Both projections come from one lineage SELECT: model_history is - # alternation-repaired for LIVE REPLAY (heals a durable ``user;user`` violation - # once instead of on every request); display_history is the full lineage - # verbatim for _display_resumed_history() (matches startup --resume). + # Both projections come from one lineage SELECT: model_history is alternation-repaired + # for LIVE REPLAY (heals a durable ``user;user`` violation once instead of on every + # request); display_history is the full lineage verbatim for _display_resumed_history() + # (matches startup --resume). model_history, display_history = self._session_db.get_resume_conversations(target_id) self.conversation_history = [m for m in (model_history or []) if m.get("role") != "session_meta"] self._resume_display_history = [m for m in (display_history or []) if m.get("role") != "session_meta"] - - # Re-open the target session so it's not marked as ended - try: + with suppress(Exception): # re-open the target session so it's not marked as ended self._session_db.reopen_session(target_id) - except Exception: - pass - _sync_agent_to_session(self, target_id, parent_session_id=old_session_id, reason="resume") - title_part = f" \"{session_meta['title']}\"" if session_meta.get("title") else "" from agent.context_compressor import is_user_originated_turn - # Count only user-originated turns: legacy compaction handoffs are durable - # role=user rows without display_kind. + # Count only user-originated turns: legacy compaction handoffs are durable role=user rows + # without display_kind. msg_count = len([m for m in self._resume_display_history if is_user_originated_turn(m)]) if self.conversation_history: - _cprint( - f" ↻ Resumed session {target_id}{title_part}" - f" ({msg_count} user message{'s' if msg_count != 1 else ''}," - f" {len(self.conversation_history)} total)" - ) + _cp(f" ↻ Resumed session {target_id}{title_part}" + f" ({_plural(msg_count, 'user message')}, {len(self.conversation_history)} total)") self._display_resumed_history() else: - _cprint(f" ↻ Resumed session {target_id}{title_part} — no messages, starting fresh.") + _cp(f" ↻ Resumed session {target_id}{title_part} — no messages, starting fresh.") - # Same contract as a startup --resume: retarget the process/tool cwd (else the - # terminal tools keep operating in the wrong repo), restore the target session's - # persisted YOLO bypass (the previous session's stops applying — the approval - # session key changed), and its model/provider (else it reverts to config default). + # Same contract as a startup --resume: retarget the process/tool cwd (else the terminal + # tools keep operating in the wrong repo), restore the target session's persisted YOLO + # bypass (the approval session key changed), and its model/provider (else it reverts to + # config default). self._restore_session_cwd(session_meta) self._restore_session_yolo(session_meta) self._restore_session_model(session_meta) @@ -1534,186 +1430,32 @@ class CLICommandsMixin: Bare/``list`` prints the same recent-sessions table /resume shows; an explicit target delegates to the resume flow so ``/sessions `` and ``/resume `` behave identically. (The TUI has a picker overlay; the classic CLI prints inline.)""" - from cli import _cprint - parts = cmd_original.split(None, 1) - arg = parts[1].strip() if len(parts) > 1 else "" - sub = arg.lower() - - # Bare /sessions or /sessions list — show recent sessions inline. - if not arg or sub in {"list", "ls", "browse"}: - if not self._session_db: - from hermes_state import format_session_db_unavailable - _cprint(f" {format_session_db_unavailable()}") - return - if not self._show_recent_sessions(reason="sessions"): - _cprint(" (._.) No previous sessions yet.") - return - - # /sessions behaves the same as /resume . - self._handle_resume_command(f"/resume {arg}") - - def _handle_worktree_command(self, cmd_original: str) -> None: - """Handle /worktree — inspect, create, or reclaim isolated git worktrees. - - Syntax: - /worktree — show the active worktree (if any) - /worktree new [name] — create a worktree and move this session into it - /worktree list — list worktrees under the repo's .worktrees/ - /worktree prune [--dry-run] — reclaim safe trees + merged branches - - ``new`` retargets the terminal/file tools (``TERMINAL_CWD`` + process cwd) at the new - tree; the launcher's exit cleanup applies (kept only with unpushed commits, as ``hermes -w``). - ``prune`` is the attended reclaim from hermes_cli/worktree_gc.py: never deletes tracked - changes, unique unpushed commits, or in-use trees.""" - import subprocess - - import cli as _cli - - parts = cmd_original.split(None, 2) - sub = parts[1].lower() if len(parts) > 1 else "" - - repo_root = _cli._git_repo_root() - - if not sub or sub in {"status", "show"}: - active = _cli._active_worktree - if active: - print(f" Active worktree: {active['path']}") - print(f" Branch: {active['branch']}") - else: - print(" No active worktree for this session.") - if repo_root: - print(" /worktree new [name] — create one and move this session into it") - print(" /worktree prune — reclaim stale trees and merged branches") - else: - print(" (not inside a git repository)") - return - - if sub in {"prune", "gc", "clean"}: - if not repo_root: - print(" Not inside a git repository.") - return - rest = parts[2].strip().lower() if len(parts) > 2 else "" - dry_run = "--dry-run" in rest or "-n" in rest.split() - from hermes_cli import worktree_gc - - active = _cli._active_worktree - tree_records = worktree_gc.audit_worktrees(repo_root, with_sizes=False) - if active: - # Never reap the tree this session is sitting in, even if judged clean+merged. - active_path = str(active.get("path") or "") - tree_records = [ - record for record in tree_records - if record.path != active_path - ] - actions = worktree_gc.reclaim_worktrees( - repo_root, dry_run=dry_run, records=tree_records - ) - actions += worktree_gc.reclaim_branches(repo_root, dry_run=dry_run) - if actions: - for line in actions: - print(f" {line}") - print(f" {len(actions)} action(s) {'planned' if dry_run else 'done'}.") - else: - print(" Nothing to reclaim — remaining trees/branches carry real work.") - kept = [ - record for record in tree_records - if record.verdict == "keep" - and "kanban" not in record.reason and "in use" not in record.reason - ] - if kept: - print(f" Preserved {len(kept)} tree(s) with real work:") - for record in kept: - print(f" {record.name}: {record.reason}") - return - - if sub in {"list", "ls"}: - if not repo_root: - print(" Not inside a git repository.") - return - try: - result = subprocess.run( - ["git", "worktree", "list"], - capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=10, cwd=repo_root, - ) - out = result.stdout.strip() if result.returncode == 0 else "" - except Exception: - out = "" - if out: - for line in out.splitlines(): - print(f" {line}") - else: - print(" Could not list worktrees.") - return - - if sub in {"new", "add", "create"}: - if not repo_root: - print(" ❌ /worktree new requires being inside a git repository.") - return - name = parts[2].strip() if len(parts) > 2 else None - from hermes_cli.config import load_config - try: - sync_base = bool(load_config().get("worktree_sync", True)) - except Exception: - sync_base = True - wt_info = _cli._setup_worktree( - repo_root=repo_root, sync_base=sync_base, name=name, - ) - if not wt_info: - return # _setup_worktree already printed the failure - # Retarget the session's terminal/file tools at the new tree (as `hermes -w` does). - try: - os.chdir(wt_info["path"]) - except OSError as e: - print(f" ⚠ Created worktree but could not enter it: {e}") - os.environ["TERMINAL_CWD"] = wt_info["path"] - # Same keep-if-unpushed cleanup as `hermes -w`. Only one tree is "active" per - # process; an earlier one keeps its own atexit registration (explicit info arg). - import atexit - _cli._active_worktree = wt_info - atexit.register(_cli._cleanup_worktree, wt_info) - print(f" ✅ Worktree ready: {wt_info['path']}") - print(f" Branch: {wt_info['branch']}") - print(" Terminal and file tools now operate in the worktree.") - return - - print(f" Unknown /worktree subcommand: {sub}") - print(" Usage: /worktree [new [name] | list]") + arg = _command_arg(cmd_original) + if arg and arg.lower() not in {"list", "ls", "browse"}: + self._handle_resume_command(f"/resume {arg}") + elif not self._session_db: + _cp(_db_unavailable_line()) + elif not self._show_recent_sessions(reason="sessions"): + _cp(" (._.) No previous sessions yet.") def _handle_branch_command(self, cmd_original: str) -> None: """Handle /branch [name] — fork the current session into a new independent copy of the full history so a different approach can be explored without losing the original.""" - from cli import _cprint, _sync_process_session_id + from cli import _sync_process_session_id if not self.conversation_history: - _cprint(" No conversation to branch — send a message first.") + _cp(" No conversation to branch — send a message first.") return - if not self._session_db: - from hermes_state import format_session_db_unavailable - _cprint(f" {format_session_db_unavailable()}") + _cp(_db_unavailable_line()) return - - parts = cmd_original.split(None, 1) - branch_name = parts[1].strip() if len(parts) > 1 else "" - - # Generate the new session ID + branch_name = _command_arg(cmd_original) now = datetime.now() - timestamp_str = now.strftime("%Y%m%d_%H%M%S") - short_uuid = uuid.uuid4().hex[:6] - new_session_id = f"{timestamp_str}_{short_uuid}" - - # Determine branch title + new_session_id = f"{now.strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}" if branch_name: branch_title = branch_name - else: - # Auto-generate from the current session title - current_title = None - if self._session_db: - current_title = self._session_db.get_session_title(self.session_id) - base = current_title or "branch" + else: # auto-generate from the current session title + base = self._session_db.get_session_title(self.session_id) or "branch" branch_title = self._session_db.get_next_title_in_lineage(base) - - # Save the current session's state before branching parent_session_id = self.session_id _end_current_session(self, "branched") @@ -1726,18 +1468,17 @@ class CLICommandsMixin: source=os.environ.get("HERMES_SESSION_SOURCE", "cli"), model=self.model, model_config={ - "max_iterations": self.max_turns, - "reasoning_config": self.reasoning_config, - "_branched_from": parent_session_id, - }, - parent_session_id=parent_session_id, - ) + "max_iterations": self.max_turns, "reasoning_config": self.reasoning_config, + "_branched_from": parent_session_id}, + parent_session_id=parent_session_id) except Exception as e: - _cprint(f" Failed to create branch session: {e}") + _cp(f" Failed to create branch session: {e}") return - # Copy history in bounded-chunk transactions. Best-effort: a failed copy still yields a usable branch. - try: + # Copy history in bounded-chunk transactions. Best-effort: a failed copy still yields a + # usable branch. api_content sidecar: the branch's first turn replays the parent's exact + # wire bytes (warm prompt cache) instead of a cold prefill. + with suppress(Exception): self._session_db.append_messages_batch( new_session_id, [ @@ -1751,23 +1492,12 @@ class CLICommandsMixin: "reasoning_details": msg.get("reasoning_details"), "codex_reasoning_items": msg.get("codex_reasoning_items"), "codex_message_items": msg.get("codex_message_items"), - # api_content sidecar: the branch's first turn replays the parent's exact - # wire bytes (warm prompt cache) instead of a cold prefill. "api_content": extract_api_content_sidecar(msg), - "timestamp": msg.get("timestamp"), - } - for msg in self.conversation_history - ], - chunk_rows=500, - ) - except Exception: - pass # Best-effort copy - - # Set title on the branch - try: + "timestamp": msg.get("timestamp")} + for msg in self.conversation_history], + chunk_rows=500) + with suppress(Exception): self._session_db.set_session_title(new_session_id, branch_title) - except Exception: - pass # Switch to the new session self._transfer_session_yolo(self.session_id, new_session_id) @@ -1776,96 +1506,173 @@ class CLICommandsMixin: self._pending_title = None self._resumed = True # Prevents auto-title generation _sync_process_session_id(new_session_id) - - # Sync the agent if self.agent: self.agent.session_start = now _sync_agent_to_session(self, new_session_id, parent_session_id=parent_session_id, reason="branch") - msg_count = len([m for m in self.conversation_history if m.get("role") == "user"]) - _cprint( - f" ⑂ Branched session \"{branch_title}\"" - f" ({msg_count} user message{'s' if msg_count != 1 else ''})" - ) - _cprint(f" Original session: {parent_session_id}") - _cprint(f" Branch session: {new_session_id}") + _cp(f" ⑂ Branched session \"{branch_title}\" ({_plural(msg_count, 'user message')})", + f" Original session: {parent_session_id}", f" Branch session: {new_session_id}") + + # ---- /worktree ------------------------------------------------------------------------ + + def _handle_worktree_command(self, cmd_original: str) -> None: + """Handle /worktree — inspect, create, or reclaim isolated git worktrees. + + ``/worktree`` shows the active worktree; ``new [name]`` creates one and moves this session + into it (retargets ``TERMINAL_CWD`` + process cwd; the launcher's exit cleanup applies — + kept only with unpushed commits, as ``hermes -w``); ``list`` lists trees under the repo's + .worktrees/; ``prune [--dry-run]`` is the attended reclaim from hermes_cli/worktree_gc.py + (never deletes tracked changes, unique unpushed commits, or in-use trees).""" + import cli as _cli + parts = cmd_original.split(None, 2) + sub = parts[1].lower() if len(parts) > 1 else "" + rest = parts[2].strip() if len(parts) > 2 else "" + repo_root = _cli._git_repo_root() + if not sub or sub in {"status", "show"}: + active = _cli._active_worktree + if active: + _pr(f" Active worktree: {active['path']}", f" Branch: {active['branch']}") + else: + print(" No active worktree for this session.") + if repo_root: + _pr(" /worktree new [name] — create one and move this session into it", + " /worktree prune — reclaim stale trees and merged branches") + else: + print(" (not inside a git repository)") + return + handler = { + "prune": self._worktree_prune, "gc": self._worktree_prune, "clean": self._worktree_prune, + "list": self._worktree_list, "ls": self._worktree_list, + "new": self._worktree_new, "add": self._worktree_new, "create": self._worktree_new, + }.get(sub) + if handler is None: + _pr(f" Unknown /worktree subcommand: {sub}", " Usage: /worktree [new [name] | list]") + return + if not repo_root: + print(" ❌ /worktree new requires being inside a git repository." + if handler is self._worktree_new else " Not inside a git repository.") + return + handler(repo_root, rest) + + def _worktree_prune(self, repo_root: str, rest: str) -> None: + import cli as _cli + from hermes_cli import worktree_gc + rest = rest.lower() + dry_run = "--dry-run" in rest or "-n" in rest.split() + active = _cli._active_worktree + tree_records = worktree_gc.audit_worktrees(repo_root, with_sizes=False) + if active: + # Never reap the tree this session is sitting in, even if judged clean+merged. + active_path = str(active.get("path") or "") + tree_records = [record for record in tree_records if record.path != active_path] + actions = worktree_gc.reclaim_worktrees(repo_root, dry_run=dry_run, records=tree_records) + actions += worktree_gc.reclaim_branches(repo_root, dry_run=dry_run) + if actions: + for line in actions: + print(f" {line}") + print(f" {len(actions)} action(s) {'planned' if dry_run else 'done'}.") + else: + print(" Nothing to reclaim — remaining trees/branches carry real work.") + kept = [ + record for record in tree_records + if record.verdict == "keep" and "kanban" not in record.reason and "in use" not in record.reason + ] + if kept: + print(f" Preserved {len(kept)} tree(s) with real work:") + for record in kept: + print(f" {record.name}: {record.reason}") + + def _worktree_list(self, repo_root: str, rest: str) -> None: + import subprocess + try: + result = subprocess.run( + ["git", "worktree", "list"], capture_output=True, text=True, encoding="utf-8", + errors="replace", timeout=10, cwd=repo_root) + out = result.stdout.strip() if result.returncode == 0 else "" + except Exception: + out = "" + if out: + for line in out.splitlines(): + print(f" {line}") + else: + print(" Could not list worktrees.") + + def _worktree_new(self, repo_root: str, rest: str) -> None: + import atexit + import cli as _cli + from hermes_cli.config import load_config + try: + sync_base = bool(load_config().get("worktree_sync", True)) + except Exception: + sync_base = True + wt_info = _cli._setup_worktree(repo_root=repo_root, sync_base=sync_base, name=rest or None) + if not wt_info: + return # _setup_worktree already printed the failure + # Retarget the session's terminal/file tools at the new tree (as `hermes -w` does). + try: + os.chdir(wt_info["path"]) + except OSError as e: + print(f" ⚠ Created worktree but could not enter it: {e}") + os.environ["TERMINAL_CWD"] = wt_info["path"] + # Same keep-if-unpushed cleanup as `hermes -w`. Only one tree is "active" per process; + # an earlier one keeps its own atexit registration (explicit info arg). + _cli._active_worktree = wt_info + atexit.register(_cli._cleanup_worktree, wt_info) + _pr(f" ✅ Worktree ready: {wt_info['path']}", f" Branch: {wt_info['branch']}", + " Terminal and file tools now operate in the worktree.") + + # ---- /personality, /pet, /hatch ------------------------------------------------------- def _handle_personality_command(self, cmd: str): - """Handle the /personality command to set predefined personalities. - - All resolution/persistence goes through hermes_cli.personality — - the single owner of personality state on every surface. - """ + """Handle /personality [name] — list or set a predefined personality. All resolution and + persistence goes through hermes_cli.personality, the single owner of personality state.""" from hermes_cli.personality import ( - describe_personality, - normalize_personality_name, - persist_personality, - prompt_text, - resolve_personality, - ) - parts = cmd.split(maxsplit=1) - - if len(parts) > 1: - # Set personality - personality_name = parts[1].strip() - - try: - name, personality_prompt = resolve_personality( - personality_name, getattr(self, "config", None) - ) - except ValueError: - print(f"(._.) Unknown personality: {personality_name.lower()}") - print(f" Available: none, {', '.join(self.personalities.keys())}") - return - - saved = persist_personality(name) - if not name: - # Neutral reset — fall back to the user-owned manual prompt. - try: - from hermes_cli.config import cfg_get, read_raw_config - - self.system_prompt = prompt_text( - cfg_get(read_raw_config(), "agent", "system_prompt", default="") - ) - except Exception: - self.system_prompt = "" - self.agent = None # Force re-init - if saved: - print("(^_^)b Personality cleared (saved to config)") - else: - print("(^_^) Personality cleared (session only)") - print(" No personality overlay — using base agent behavior.") - else: - self.system_prompt = personality_prompt - self.agent = None # Force re-init - if saved: - print(f"(^_^)b Personality set to '{name}' (saved to config)") - else: - print(f"(^_^) Personality set to '{name}' (session only)") - print(f" \"{personality_prompt[:60]}{'...' if len(personality_prompt) > 60 else ''}\"") - else: - # Show available personalities + describe_personality, normalize_personality_name, persist_personality, prompt_text, + resolve_personality) + personality_name = _command_arg(cmd) + if not personality_name: try: from hermes_cli.config import read_raw_config - current = normalize_personality_name( - (read_raw_config().get("display") or {}).get("personality", "") - ) + (read_raw_config().get("display") or {}).get("personality", "")) except Exception: current = "" print() - print("+" + "-" * 50 + "+") - print("|" + " " * 12 + "(^o^)/ Personalities" + " " * 15 + "|") - print("+" + "-" * 50 + "+") + _pr("+" + "-" * 50 + "+", "|" + " " * 12 + "(^o^)/ Personalities" + " " * 15 + "|", + "+" + "-" * 50 + "+") print() marker = " *" if not current else " " print(f" {marker}{'none':<12} - (no personality overlay)") for name, prompt in self.personalities.items(): marker = " *" if name == current else " " print(f" {marker}{name:<12} - {describe_personality(prompt)}") - print() - print(" Usage: /personality (* = active)") - print() + _pr("", " Usage: /personality (* = active)", "") + return + try: + name, personality_prompt = resolve_personality(personality_name, getattr(self, "config", None)) + except ValueError: + print(f"(._.) Unknown personality: {personality_name.lower()}") + print(f" Available: none, {', '.join(self.personalities.keys())}") + return + saved = persist_personality(name) + scope = "(saved to config)" if saved else "(session only)" + face = "(^_^)b" if saved else "(^_^)" + if not name: + # Neutral reset — fall back to the user-owned manual prompt. + try: + from hermes_cli.config import cfg_get, read_raw_config + self.system_prompt = prompt_text( + cfg_get(read_raw_config(), "agent", "system_prompt", default="")) + except Exception: + self.system_prompt = "" + self.agent = None # Force re-init + _pr(f"{face} Personality cleared {scope}", + " No personality overlay — using base agent behavior.") + else: + self.system_prompt = personality_prompt + self.agent = None # Force re-init + _pr(f"{face} Personality set to '{name}' {scope}", + f" \"{_ellipsize(personality_prompt, 60)}\"") def _handle_pet_command(self, cmd: str): """Toggle, browse, or adopt a petdex mascot. @@ -1875,49 +1682,39 @@ class CLICommandsMixin: Writes ``display.pet.*`` to config; pet surfaces pick it up on their next poll.""" from agent.pet import store from agent.pet.manifest import ManifestError - from hermes_cli.pets import _set_active, _set_enabled, print_pet_gallery, set_pet_scale, toggle_pet_display - - parts = cmd.split(maxsplit=1) - arg = parts[1].strip() if len(parts) > 1 else "" + from hermes_cli.pets import ( + _set_active, _set_enabled, print_pet_gallery, set_pet_scale, toggle_pet_display) + arg = _command_arg(cmd) low = arg.lower() - if not arg or low == "toggle": enabled, name, err = toggle_pet_display() if err: print(f"(x_x) {err}") - return - if enabled: + elif enabled: print(f"(^_^)b {name} is out — it'll pop in shortly.") else: print(f"(-_-)zzZ {name} put away." if name else "(-_-)zzZ Pet put away.") - return - - if low in ("list", "gallery", "browse", "all"): + elif low in ("list", "gallery", "browse", "all"): print_pet_gallery() - return - - if low == "scale" or low.startswith("scale "): + elif low == "scale" or low.startswith("scale "): value = arg[len("scale"):].strip() if not value: print("(o_o) Usage: /pet scale (e.g. /pet scale 0.5)") return scale, err = set_pet_scale(value) print(f"(x_x) {err}" if err else f"(^_^) Pet scale → {scale:g}.") - return - - if low == "off": + elif low == "off": _set_enabled(False) print("(-_-)zzZ Pet put away.") - return - - print(f"(o_o) Fetching '{arg}' from petdex…") - try: - pet = store.install_pet(arg) - except (store.PetStoreError, ManifestError) as exc: - print(f"(x_x) Couldn't adopt '{arg}': {exc}") - return - _set_active(arg) - print(f"(^_^)b {pet.display_name} is out — it'll pop in shortly.") + else: + print(f"(o_o) Fetching '{arg}' from petdex…") + try: + pet = store.install_pet(arg) + except (store.PetStoreError, ManifestError) as exc: + print(f"(x_x) Couldn't adopt '{arg}': {exc}") + return + _set_active(arg) + print(f"(^_^)b {pet.display_name} is out — it'll pop in shortly.") def _handle_hatch_command(self, cmd: str): """Generate ("hatch") a new petdex pet from a description: base look, one animation row @@ -1927,14 +1724,11 @@ class CLICommandsMixin: from agent.pet.generate import orchestrate from agent.pet.generate.imagegen import GenerationError from hermes_cli.pets import _set_active - - parts = cmd.split(maxsplit=1) - concept = parts[1].strip() if len(parts) > 1 else "" - + concept = _command_arg(cmd) if not concept: - # Dispatched from the process_loop daemon thread while prompt_toolkit owns - # stdin — a raw input() here never renders and eats keystrokes. Use the - # thread-aware prompt helper (run_in_terminal; None when prompting isn't safe). + # Dispatched from the process_loop daemon thread while prompt_toolkit owns stdin — a + # raw input() here never renders and eats keystrokes. Use the thread-aware prompt + # helper (run_in_terminal; None when prompting isn't safe). prompt_helper = getattr(self, "_prompt_text_input", None) if callable(prompt_helper): concept = (prompt_helper("(o_o) Describe your pet: ") or "").strip() @@ -1944,7 +1738,6 @@ class CLICommandsMixin: except (EOFError, KeyboardInterrupt): print() return - if not concept: print("(o_o) Usage: /hatch (e.g. /hatch a tiny cyber fox)") return @@ -1952,47 +1745,38 @@ class CLICommandsMixin: # A short, friendly display name from the first few words of the concept. display_name = " ".join(w.capitalize() for w in concept.split()[:3])[:28].strip() or "Pet" slug = store.slugify(display_name) or store.slugify(concept) or "pet" - print(f"(o_o) Designing '{concept}'… (a minute of image-model calls)") try: drafts = orchestrate.generate_base_drafts(concept, n=1) except GenerationError as exc: print(f"(x_x) Couldn't generate a base look: {exc}") return - if not drafts: print("(x_x) No base draft came back — try again.") return def _progress(event: str, detail: str) -> None: - if event == "row": - # detail is "::"; show the state name. - state = detail.split(":", 1)[0] - print(f" ┊ drawing {state}…") + if event == "row": # detail is "::"; show the state name. + print(f" ┊ drawing {detail.split(':', 1)[0]}…") elif event == "compose": print(" ┊ composing spritesheet…") elif event == "save": print(" ┊ saving…") - try: result = orchestrate.hatch_pet( - base_image=drafts[0], - slug=slug, - display_name=display_name, - concept=concept, - on_progress=_progress, - ) + base_image=drafts[0], slug=slug, display_name=display_name, concept=concept, + on_progress=_progress) except GenerationError as exc: print(f"(x_x) Hatch failed: {exc}") return - _set_active(result.slug) print(f"(^_^)b {result.display_name} hatched and adopted — it'll pop in shortly!") + # ---- /cron ---------------------------------------------------------------------------- + def _handle_cron_command(self, cmd: str): """Handle the /cron command to manage scheduled tasks.""" import shlex - tokens = shlex.split(cmd) if len(tokens) == 1: self._cron_overview() @@ -2003,37 +1787,30 @@ class CLICommandsMixin: return handler = _CRON_SUBCOMMANDS.get(subcommand) if handler is None: - print(f"(._.) Unknown cron command: {subcommand}") - print(" Available: list, add, edit, pause, resume, run, remove") + _pr(f"(._.) Unknown cron command: {subcommand}", + " Available: list, add, edit, pause, resume, run, remove") return - handler(self, subcommand, opts) + getattr(self, handler)(subcommand, opts) def _cron_overview(self) -> None: print() - print("+" + "-" * 68 + "+") - print("|" + " " * 22 + "(^_^) Scheduled Tasks" + " " * 23 + "|") - print("+" + "-" * 68 + "+") + _pr("+" + "-" * 68 + "+", "|" + " " * 22 + "(^_^) Scheduled Tasks" + " " * 23 + "|", + "+" + "-" * 68 + "+") print() - print(" Commands:") - print(" /cron list") - print(' /cron add "every 2h" "Check server status" [--skill blogwatcher]') - print(' /cron edit --schedule "every 4h" --prompt "New task"') - print(" /cron edit --skill blogwatcher --skill maps") - print(" /cron edit --remove-skill blogwatcher") - print(" /cron edit --clear-skills") - print(" /cron pause ") - print(" /cron resume ") - print(" /cron run ") - print(" /cron remove ") + _pr(" Commands:", " /cron list", + ' /cron add "every 2h" "Check server status" [--skill blogwatcher]', + ' /cron edit --schedule "every 4h" --prompt "New task"', + " /cron edit --skill blogwatcher --skill maps", + " /cron edit --remove-skill blogwatcher", + " /cron edit --clear-skills", " /cron pause ", + " /cron resume ", " /cron run ", " /cron remove ") print() result = _cron_api(action="list") jobs = result.get("jobs", []) if result.get("success") else [] if jobs: - print(" Current Jobs:") - print(" " + "-" * 63) + _pr(" Current Jobs:", " " + "-" * 63) for job in jobs: - repeat_str = job.get("repeat", "?") - print(f" {job['job_id'][:12]:<12} | {job['schedule']:<15} | {repeat_str:<8}") + print(f" {job['job_id'][:12]:<12} | {job['schedule']:<15} | {job.get('repeat', '?'):<8}") if job.get("skills"): print(f" Skills: {', '.join(job['skills'])}") print(f" {job.get('prompt_preview', '')}") @@ -2051,14 +1828,12 @@ class CLICommandsMixin: print("(._.) No scheduled jobs.") return print() - print("Scheduled Jobs:") - print("-" * 80) + _pr("Scheduled Jobs:", "-" * 80) for job in jobs: - print(f" ID: {job['job_id']}") - print(f" Name: {job['name']}") - print(f" State: {job.get('state', '?')}") - print(f" Schedule: {job['schedule']} ({job.get('repeat', '?')})") - print(f" Next run: {job.get('next_run_at', 'N/A')}") + _pr(f" ID: {job['job_id']}", f" Name: {job['name']}", + f" State: {job.get('state', '?')}", + f" Schedule: {job['schedule']} ({job.get('repeat', '?')})", + f" Next run: {job.get('next_run_at', 'N/A')}") if job.get("skills"): print(f" Skills: {', '.join(job['skills'])}") print(f" Prompt: {job.get('prompt_preview', '')}") @@ -2084,20 +1859,17 @@ class CLICommandsMixin: return result = _cron_api( action="create", schedule=schedule, prompt=prompt or None, name=opts["name"], - deliver=opts["deliver"], repeat=opts["repeat"], skills=skills or None, - ) - if result.get("success"): - print(f"(^_^)b Created job: {result['job_id']}") - print(f" Schedule: {result['schedule']}") - if result.get("skills"): - print(f" Skills: {', '.join(result['skills'])}") - print(f" Next run: {result['next_run_at']}") - else: + deliver=opts["deliver"], repeat=opts["repeat"], skills=skills or None) + if not result.get("success"): print(f"(x_x) Failed to create job: {result.get('error')}") + return + _pr(f"(^_^)b Created job: {result['job_id']}", f" Schedule: {result['schedule']}") + if result.get("skills"): + print(f" Skills: {', '.join(result['skills'])}") + print(f" Next run: {result['next_run_at']}") def _cron_edit(self, subcommand: str, opts: dict) -> None: from cli import get_job - positionals = opts["positionals"] if not positionals: print("(._.) Usage: /cron edit [--schedule ...] [--prompt ...] [--skill ...]") @@ -2113,31 +1885,24 @@ class CLICommandsMixin: replacement_skills = _normalize_skills(opts["skills"]) add_skills = _normalize_skills(opts["add_skills"]) remove_skills = set(_normalize_skills(opts["remove_skills"])) - existing_skills = list(existing.get("skills") or ([] if not existing.get("skill") else [existing.get("skill")])) if opts["clear_skills"]: final_skills = [] elif replacement_skills: final_skills = replacement_skills elif add_skills or remove_skills: + existing_skills = list( + existing.get("skills") or ([existing["skill"]] if existing.get("skill") else [])) final_skills = [skill for skill in existing_skills if skill not in remove_skills] - for skill in add_skills: - if skill not in final_skills: - final_skills.append(skill) - + final_skills += [skill for skill in add_skills if skill not in final_skills] result = _cron_api( action="update", job_id=job_id, schedule=opts["schedule"], prompt=opts["prompt"], - name=opts["name"], deliver=opts["deliver"], repeat=opts["repeat"], skills=final_skills, - ) - if result.get("success"): - job = result["job"] - print(f"(^_^)b Updated job: {job['job_id']}") - print(f" Schedule: {job['schedule']}") - if job.get("skills"): - print(f" Skills: {', '.join(job['skills'])}") - else: - print(" Skills: none") - else: + name=opts["name"], deliver=opts["deliver"], repeat=opts["repeat"], skills=final_skills) + if not result.get("success"): print(f"(x_x) Failed to update job: {result.get('error')}") + return + job = result["job"] + _pr(f"(^_^)b Updated job: {job['job_id']}", f" Schedule: {job['schedule']}", + f" Skills: {', '.join(job['skills'])}" if job.get("skills") else " Skills: none") def _cron_job_action(self, subcommand: str, opts: dict) -> None: """pause / resume / run / remove (aliases rm, delete) on one job id.""" @@ -2147,32 +1912,28 @@ class CLICommandsMixin: return job_id = positionals[0] action = "remove" if subcommand in {"remove", "rm", "delete"} else subcommand - result = _cron_api(action=action, job_id=job_id, reason="paused from /cron" if action == "pause" else None) + result = _cron_api(action=action, job_id=job_id, + reason="paused from /cron" if action == "pause" else None) if not result.get("success"): print(f"(x_x) Failed to {action} job: {result.get('error')}") return - if action == "pause": - print(f"(^_^)b Paused job: {result['job']['name']} ({job_id})") - elif action == "resume": - print(f"(^_^)b Resumed job: {result['job']['name']} ({job_id})") - print(f" Next run: {result['job'].get('next_run_at')}") - elif action == "run": - print(f"(^_^)b Triggered job: {result['job']['name']} ({job_id})") - print(" It will run on the next scheduler tick.") - else: + if action == "remove": removed = result.get("removed_job", {}) print(f"(^_^)b Removed job: {removed.get('name', job_id)} ({job_id})") + return + verb = {"pause": "Paused", "resume": "Resumed", "run": "Triggered"}[action] + print(f"(^_^)b {verb} job: {result['job']['name']} ({job_id})") + if action == "resume": + print(f" Next run: {result['job'].get('next_run_at')}") + elif action == "run": + print(" It will run on the next scheduler tick.") + + # ---- delegating handlers: /suggestions, /blueprint, /curator, /kanban, /skills, /memory -- def _handle_suggestions_command(self, cmd: str): """Handle /suggestions — review/accept/dismiss suggested automations via the shared handler. CLI origin is the local platform so an accepted job's "origin" delivery resolves to a home channel.""" - import shlex - - try: - tokens = shlex.split(cmd)[1:] if cmd else [] - except ValueError: - tokens = (cmd or "").split()[1:] - args = " ".join(tokens) + args = " ".join(_shlex_args(cmd)) try: from hermes_cli.suggestions_cmd import handle_suggestions_command output = handle_suggestions_command(args) @@ -2185,12 +1946,7 @@ class CLICommandsMixin: Bare lists the catalog; ```` seeds the agent to ask for each value conversationally (``agent_seed``, run as the next turn); `` slot=val …`` creates the job directly.""" import shlex - - try: - tokens = shlex.split(cmd)[1:] if cmd else [] - except ValueError: - tokens = (cmd or "").split()[1:] - args = " ".join(shlex.quote(t) for t in tokens) + args = " ".join(shlex.quote(t) for t in _shlex_args(cmd)) try: from hermes_cli.blueprint_cmd import handle_blueprint_command result = handle_blueprint_command(args) @@ -2200,39 +1956,27 @@ class CLICommandsMixin: self._console_print(result.text) seed = getattr(result, "agent_seed", None) if seed: - # One-shot: the interactive loop picks this up right after the - # slash command returns and runs it as a normal agent turn. + # One-shot: the interactive loop picks this up right after the slash command + # returns and runs it as a normal agent turn. self._pending_agent_seed = seed def _handle_curator_command(self, cmd: str): - """Handle /curator slash command. - - Delegates to hermes_cli.curator so the CLI and the `hermes curator` - subcommand share the same handler set. - """ + """Handle /curator — delegates to hermes_cli.curator so the CLI and the `hermes curator` + subcommand share the same handler set.""" import shlex - tokens = shlex.split(cmd)[1:] if cmd else [] - if not tokens: - tokens = ["status"] - try: from hermes_cli.curator import cli_main - cli_main(tokens) + cli_main(tokens or ["status"]) except SystemExit: - # argparse calls sys.exit() on --help or errors; swallow so we - # don't kill the interactive session. - pass + pass # argparse exits on --help/errors; don't kill the interactive session except Exception as exc: print(f"(._.) curator: {exc}") def _handle_kanban_command(self, cmd: str): """Handle /kanban — strip the leading ``/kanban`` and hand the rest to ``kanban.run_slash``.""" from hermes_cli.kanban import run_slash - - rest = cmd.strip() - if rest.startswith("/"): - rest = rest.lstrip("/") + rest = cmd.strip().lstrip("/") if rest.startswith("kanban"): rest = rest[len("kanban"):].lstrip() try: @@ -2243,19 +1987,16 @@ class CLICommandsMixin: print(output) def _handle_skills_command(self, cmd: str): - """Handle /skills slash command — delegates to hermes_cli.skills_hub.""" + """Handle /skills slash command — delegates to hermes_cli.skills_hub, after intercepting the + write-approval review subcommands (pending/approve/reject/diff/mode).""" from cli import ChatConsole - # Intercept write-approval review subcommands first (pending/approve/ - # reject/diff/mode); everything else goes to the skills hub. - parts = cmd.strip().split() - args = parts[1:] if len(parts) > 1 else [] - if args and args[0].lower() in {"pending", "approve", "apply", "reject", - "deny", "drop", "diff", "approval", "mode"}: + args = cmd.strip().split()[1:] + review_words = {"pending", "approve", "apply", "reject", "deny", "drop", "diff", "approval", "mode"} + if args and args[0].lower() in review_words: from hermes_cli.write_approval_commands import handle_pending_subcommand from tools import write_approval as wa out = handle_pending_subcommand( - wa.SKILLS, args, - set_mode_fn=lambda enabled: self._save_write_approval("skills", enabled), + wa.SKILLS, args, set_mode_fn=lambda enabled: self._save_write_approval("skills", enabled), ) if out is not None: print(out) @@ -2263,6 +2004,30 @@ class CLICommandsMixin: from hermes_cli.skills_hub import handle_skills_slash handle_skills_slash(cmd, ChatConsole()) + def _handle_memory_command(self, cmd: str): + """Handle /memory slash command — pending review + approval-gate toggle.""" + from hermes_cli.write_approval_commands import handle_pending_subcommand + from tools import write_approval as wa + args = cmd.strip().split()[1:] + store = getattr(self.agent, "_memory_store", None) if getattr(self, "agent", None) else None + if store is None: + # No live agent store (e.g. /memory approve from the Desktop GUI): apply against a + # freshly loaded on-disk store, mirroring the gateway path. It persists to the same + # MEMORY/USER.md and honors the user's configured char limits. + from tools.memory_tool import load_on_disk_store + store = load_on_disk_store() + out = handle_pending_subcommand( + wa.MEMORY, args, memory_store=store, + set_mode_fn=lambda enabled: self._save_write_approval("memory", enabled)) + print(out if out is not None else + "Unknown /memory subcommand. Use: pending, approve , reject , approval .") + + def _save_write_approval(self, subsystem: str, enabled: bool): + """Persist .write_approval to config (for /memory|/skills approval).""" + _save(f"{subsystem}.write_approval", bool(enabled)) + + # ---- prompt-queueing handlers: /learn, /plan, /init ----------------------------------- + def _queue_prompt_turn(self, msg: str, command: str) -> None: """Inject ``msg`` onto the agent's input queue as the next normal user turn (the /learn, /plan, /init pattern: no engine, no model-tool footprint, prompt-cache safe).""" @@ -2272,136 +2037,75 @@ class CLICommandsMixin: print(f" {command} needs an active chat session to run.") def _handle_learn_command(self, cmd: str): - """Handle /learn — distill a reusable skill from anything the user describes. - - Open-ended: the argument is free text describing the source(s) — a directory, a URL, - "what we just did", pasted notes. The live agent gathers the material with the tools - it already has and authors the skill via ``skill_manage``.""" + """Handle /learn — distill a reusable skill from anything the user describes (a directory, + a URL, "what we just did", pasted notes). The live agent gathers the material with the + tools it already has and authors the skill via ``skill_manage``.""" from agent.learn_prompt import build_learn_prompt - user_request = _command_arg(cmd) - if user_request: - print("\n⚡ Learning a skill from what you described...") - else: - print("\n⚡ Learning a skill from this conversation...") + print("\n⚡ Learning a skill from what you described..." if user_request + else "\n⚡ Learning a skill from this conversation...") self._queue_prompt_turn(build_learn_prompt(user_request), "/learn") def _handle_plan_command(self, cmd: str): """Handle /plan — write a markdown implementation plan, no execution. The live agent inspects the workspace read-only and saves the plan under ``.hermes/plans/``.""" from agent.plan_prompt import build_plan_prompt - task = _command_arg(cmd) # optional — empty infers the task from conversation context - if task: - print(f"\n📋 Planning: {task[:80]}{'...' if len(task) > 80 else ''}") - else: - print("\n📋 Planning from this conversation's context...") + print(f"\n📋 Planning: {_ellipsize(task, 80)}" if task + else "\n📋 Planning from this conversation's context...") self._queue_prompt_turn(build_plan_prompt(task), "/plan") def _handle_init_command(self, cmd: str): """Handle /init — generate or update AGENTS.md from a project scan performed by the live agent with its own read-only tools.""" from hermes_cli.init_command import build_init_prompt_for_cwd - msg = build_init_prompt_for_cwd(extra=_command_arg(cmd)) # optional user emphasis - if "UPDATE the existing AGENTS.md" in msg: - print("\n⚡ Updating AGENTS.md from a project scan...") - else: - print("\n⚡ Generating AGENTS.md from a project scan...") + verb = "Updating" if "UPDATE the existing AGENTS.md" in msg else "Generating" + print(f"\n⚡ {verb} AGENTS.md from a project scan...") self._queue_prompt_turn(msg, "/init") - def _handle_memory_command(self, cmd: str): - """Handle /memory slash command — pending review + approval-gate toggle.""" - from hermes_cli.write_approval_commands import handle_pending_subcommand - from tools import write_approval as wa - parts = cmd.strip().split() - args = parts[1:] if len(parts) > 1 else [] - store = getattr(self.agent, "_memory_store", None) if getattr(self, "agent", None) else None - if store is None: - # No live agent store (e.g. /memory approve from the Desktop GUI): apply against - # a freshly loaded on-disk store, mirroring the gateway path. It persists to the - # same MEMORY/USER.md and honors the user's configured char limits. - from tools.memory_tool import load_on_disk_store - store = load_on_disk_store() - out = handle_pending_subcommand( - wa.MEMORY, args, - memory_store=store, - set_mode_fn=lambda enabled: self._save_write_approval("memory", enabled), - ) - if out is None: - out = ("Unknown /memory subcommand. " - "Use: pending, approve , reject , approval .") - print(out) - - def _save_write_approval(self, subsystem: str, enabled: bool): - """Persist .write_approval to config (for /memory|/skills approval).""" - from cli import save_config_value - save_config_value(f"{subsystem}.write_approval", bool(enabled)) + # ---- side-session handlers: /bg, /btw ------------------------------------------------- def _handle_background_command(self, cmd: str): """Handle /bg — run a prompt in a separate background session (its own AIAgent on a thread); the result prints here without touching the active history.""" - from cli import AIAgent, _cprint, set_approval_callback, set_secret_capture_callback, set_sudo_password_callback - parts = cmd.strip().split(maxsplit=1) - if len(parts) < 2 or not parts[1].strip(): - _cprint(" Usage: /bg ") - _cprint(" Example: /bg Summarize the top HN stories today") - _cprint(" (For a side question about this conversation, use /btw .)") - _cprint(" The task runs in a separate session and results display here when done.") + from cli import ( + AIAgent, set_approval_callback, set_secret_capture_callback, set_sudo_password_callback) + prompt = _command_arg(cmd) + if not prompt: + _cp(" Usage: /bg ", " Example: /bg Summarize the top HN stories today", + " (For a side question about this conversation, use /btw .)", + " The task runs in a separate session and results display here when done.") return - - prompt = parts[1].strip() self._background_task_counter += 1 task_num = self._background_task_counter task_id = f"bg_{datetime.now().strftime('%H%M%S')}_{uuid.uuid4().hex[:6]}" - - # Make sure we have valid credentials if not self._ensure_runtime_credentials(): - _cprint(" (>_<) Cannot start background task: no valid credentials.") + _cp(" (>_<) Cannot start background task: no valid credentials.") return - - _cprint(f" 🔄 Background task #{task_num} started: \"{prompt[:60]}{'...' if len(prompt) > 60 else ''}\"") - _cprint(f" Task ID: {task_id}") - _cprint(" You can continue chatting — results will appear when done.\n") - + preview = _ellipsize(prompt, 60) + _cp(f" 🔄 Background task #{task_num} started: \"{preview}\"", f" Task ID: {task_id}", + " You can continue chatting — results will appear when done.\n") turn_route = self._resolve_turn_agent_config(prompt) + runtime = turn_route["runtime"] def run_background(): set_sudo_password_callback(self._sudo_password_callback) set_approval_callback(self._approval_callback) - try: + with suppress(Exception): set_secret_capture_callback(self._secret_capture_callback) - except Exception: - pass try: bg_agent = AIAgent( - model=turn_route["model"], - api_key=turn_route["runtime"].get("api_key"), - base_url=turn_route["runtime"].get("base_url"), - provider=turn_route["runtime"].get("provider"), - api_mode=turn_route["runtime"].get("api_mode"), - acp_command=turn_route["runtime"].get("command"), - acp_args=turn_route["runtime"].get("args"), - max_tokens=turn_route["runtime"].get("max_tokens"), - max_iterations=self.max_turns, - enabled_toolsets=self.enabled_toolsets, - quiet_mode=True, - verbose_logging=False, - session_id=task_id, - platform="cli", - session_db=self._session_db, - reasoning_config=self.reasoning_config, + model=turn_route["model"], api_key=runtime.get("api_key"), + base_url=runtime.get("base_url"), provider=runtime.get("provider"), + api_mode=runtime.get("api_mode"), acp_command=runtime.get("command"), + acp_args=runtime.get("args"), max_tokens=runtime.get("max_tokens"), + max_iterations=self.max_turns, enabled_toolsets=self.enabled_toolsets, + quiet_mode=True, verbose_logging=False, session_id=task_id, platform="cli", + session_db=self._session_db, reasoning_config=self.reasoning_config, service_tier=self.service_tier, request_overrides=turn_route.get("request_overrides"), - providers_allowed=self._providers_only, - providers_ignored=self._providers_ignore, - providers_order=self._providers_order, - provider_sort=self._provider_sort, - provider_require_parameters=self._provider_require_params, - provider_data_collection=self._provider_data_collection, - openrouter_min_coding_score=self._openrouter_min_coding_score, - fallback_model=self._fallback_model, - ) + **{kw: getattr(self, attr) for kw, attr in _BG_PROVIDER_KWARGS.items()}) # Silence raw spinner; route thinking through TUI widget when no foreground agent is active. bg_agent._print_fn = lambda *_a, **_kw: None @@ -2411,55 +2115,34 @@ class CLICommandsMixin: self._spinner_text = text if self._app: self._app.invalidate() - bg_agent.thinking_callback = _bg_thinking - - result = bg_agent.run_conversation( - user_message=prompt, - task_id=task_id, - ) - + result = bg_agent.run_conversation(user_message=prompt, task_id=task_id) response = result.get("final_response", "") if result else "" if not response and result and result.get("error"): response = f"Error: {result['error']}" - _print_side_result_panel( self, - header_lines=[ - f" ✅ Background task #{task_num} complete", - f" Prompt: \"{prompt[:60]}{'...' if len(prompt) > 60 else ''}\"", - ], + header_lines=[f" ✅ Background task #{task_num} complete", + f" Prompt: \"{preview}\""], body=response, title_suffix=f"(background #{task_num})", - empty_note=" (No response generated)", - ) - - # Play bell if enabled + empty_note=" (No response generated)") if self.bell_on_complete: sys.stdout.write("\a") sys.stdout.flush() - except Exception as e: - # Same TUI refresh pattern as success path - if self._app: - self._app.invalidate() - time.sleep(0.05) - print() - _cprint(f" ❌ Background task #{task_num} failed: {e}") + _refresh_tui_before_print(self) + _cp(f" ❌ Background task #{task_num} failed: {e}") finally: - try: + with suppress(Exception): set_sudo_password_callback(None) set_approval_callback(None) set_secret_capture_callback(None) - except Exception: - pass self._background_tasks.pop(task_id, None) - # Clear spinner only if no foreground agent owns it - if not self._agent_running: + if not self._agent_running: # clear spinner only if no foreground agent owns it self._spinner_text = "" if self._app: self._invalidate(min_interval=0) - thread = threading.Thread(target=run_background, daemon=True, name=f"bg-task-{task_id}") self._background_tasks[task_id] = thread thread.start() @@ -2468,172 +2151,136 @@ class CLICommandsMixin: """Handle /btw — answer a side question about this conversation from a history snapshot via a one-shot auxiliary call. The live session is never touched (no history mutation, no role-alternation risk, no cache invalidation).""" - from cli import _cprint - - parts = cmd.strip().split(maxsplit=1) - if len(parts) < 2 or not parts[1].strip(): - _cprint(" Usage: /btw ") - _cprint(" Example: /btw which file was that error in?") - _cprint(" Answers a quick question about this conversation without interrupting it.") - _cprint(" (For an independent background task, use /bg .)") + question = _command_arg(cmd) + if not question: + _cp(" Usage: /btw ", " Example: /btw which file was that error in?", + " Answers a quick question about this conversation without interrupting it.", + " (For an independent background task, use /bg .)") return - - question = parts[1].strip() - if not self._ensure_runtime_credentials(): - _cprint(" (>_<) Cannot answer side question: no valid credentials.") + _cp(" (>_<) Cannot answer side question: no valid credentials.") return - # Snapshot NOW, on the UI thread — the foreground turn keeps appending - # to conversation_history while the worker runs. + # Snapshot NOW, on the UI thread — the foreground turn keeps appending to + # conversation_history while the worker runs. history_snapshot = list(self.conversation_history or []) # Live agent → cache-parity fork (full context, warm cache reads). parent_agent = self.agent turn_route = self._resolve_turn_agent_config(question) + runtime = turn_route["runtime"] main_runtime = { "model": turn_route["model"], - "provider": turn_route["runtime"].get("provider"), - "base_url": turn_route["runtime"].get("base_url"), - "api_key": turn_route["runtime"].get("api_key"), - "api_mode": turn_route["runtime"].get("api_mode"), - } - - preview = question[:60] + ("..." if len(question) > 60 else "") - _cprint(f" 💬 Side question: \"{preview}\"") - _cprint(" Answering from a snapshot of this conversation — the current work continues.\n") + **{k: runtime.get(k) for k in ("provider", "base_url", "api_key", "api_mode")}} + preview = _ellipsize(question, 60) + _cp(f" 💬 Side question: \"{preview}\"", + " Answering from a snapshot of this conversation — the current work continues.\n") def run_side_question(): try: from agent.side_question import answer_side_question answer = answer_side_question( - question, - history_snapshot, - parent_agent=parent_agent, - main_runtime=main_runtime, + question, history_snapshot, parent_agent=parent_agent, main_runtime=main_runtime, ) _print_side_result_panel( - self, - header_lines=[f" 💬 /btw: \"{preview}\""], - body=answer, - title_suffix="(btw)", - empty_note=" (No answer generated)", - ) + self, header_lines=[f" 💬 /btw: \"{preview}\""], body=answer, + title_suffix="(btw)", empty_note=" (No answer generated)") except Exception as e: - if self._app: - self._app.invalidate() - time.sleep(0.05) - print() - _cprint(f" ❌ /btw failed: {e}") + _refresh_tui_before_print(self) + _cp(f" ❌ /btw failed: {e}") finally: if self._app: self._invalidate(min_interval=0) - threading.Thread(target=run_side_question, daemon=True, name="btw-side-question").start() + # ---- /bundles, /browser --------------------------------------------------------------- + def _handle_bundles_command(self, cmd: str) -> None: """In-session ``/bundles`` — show installed skill bundles (``hermes bundles list`` rendered inside the running CLI). Bundles are loaded via ``/``.""" - from cli import ChatConsole, _BOLD, _DIM, _RST, _accent_hex, _cprint + from cli import ChatConsole, _BOLD, _RST, _accent_hex from hermes_cli.slash_exec import CommandContext, execute_command - reply = execute_command("bundles", CommandContext(surface="cli")) if "error" in reply.data: - _cprint(f"\033[1;31mBundle subsystem unavailable: {reply.data['error']}{_RST}") + _cp(f"\033[1;31mBundle subsystem unavailable: {reply.data['error']}{_RST}") return - bundles = reply.data["bundles"] if not bundles: - _cprint(" No skill bundles installed.") - _cprint( - f" {_DIM}Create one with: hermes bundles create " - f" --skill --skill {_RST}" - ) - _cprint(f" {_DIM}Directory: {reply.data['dir']}{_RST}") + _cp(" No skill bundles installed.", + _dim_line('Create one with: hermes bundles create --skill --skill '), + _dim_line(f"Directory: {reply.data['dir']}")) return - - _cprint(f"\n ▣ {_BOLD}Skill Bundles{_RST} ({len(bundles)} installed):") + _cp(f"\n ▣ {_BOLD}Skill Bundles{_RST} ({len(bundles)} installed):") for info in bundles: skill_count = len(info.get("skills", [])) desc = info.get("description") or f"Load {skill_count} skills" ChatConsole().print( f" [bold {_accent_hex()}]/{info['slug']:<20}[/] " - f"[dim]-[/] {_escape(desc)} [dim]({skill_count} skills)[/]" - ) + f"[dim]-[/] {_escape(desc)} [dim]({skill_count} skills)[/]") for s in info.get("skills", []): ChatConsole().print(f" [dim]· {_escape(s)}[/]") - _cprint( - f"\n {_DIM}Invoke a bundle with /. " - f"Manage with `hermes bundles`.{_RST}" - ) + _cp("\n" + _dim_line("Invoke a bundle with /. Manage with `hermes bundles`.")) def _handle_browser_command(self, cmd: str): """Handle /browser connect|disconnect|status|use — manage the live Chromium-family CDP connection.""" - parts = cmd.strip().split(None, 1) - sub = parts[1].lower().strip() if len(parts) > 1 else "status" + sub = _command_arg(cmd).lower() or "status" if sub == "use" or sub.startswith("use "): _browser_use(self, sub.split(None, 1)[1].strip() if " " in sub else "on") elif sub.startswith("connect"): connect_parts = cmd.strip().split(None, 2) # ["/browser", "connect", "ws://..."] - _browser_connect(self, connect_parts[2].strip() if len(connect_parts) > 2 else DEFAULT_BROWSER_CDP_URL) + url = connect_parts[2].strip() if len(connect_parts) > 2 else DEFAULT_BROWSER_CDP_URL + _browser_connect(self, url) elif sub == "disconnect": _browser_disconnect(self) elif sub == "status": _browser_status() else: _say_block( - "Usage: /browser connect|disconnect|status|use", - "", + "Usage: /browser connect|disconnect|status|use", "", " connect Connect browser tools to your live Chromium-family browser session", " disconnect Revert to default browser backend", " status Show current browser mode", - " use [off] Switch to Browser Use mode (CLI 3.0) / back to built-in tools", - ) + " use [off] Switch to Browser Use mode (CLI 3.0) / back to built-in tools") + + # ---- /heartbeat, /refine, /review ----------------------------------------------------- + + def _session_manager(self, getter, label: str): + """The session-scoped manager from ``getter()``, or None after the standard dim + "