"""Classic-TUI run loop phases: input processing, after-turn bookkeeping, startup banner/prewarm/maintenance, prompt_toolkit application build, signal handlers and shutdown. Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. """ from __future__ import annotations import errno import logging import os import queue import shutil import sys import threading import time from agent.i18n import t from agent.interrupt_compat import request_hard_interrupt from agent.pet import render as pet_render from contextlib import suppress from hermes_constants import get_hermes_home from prompt_toolkit.application import Application from rich.markup import escape as _escape # Log-record parity with the origin module. logger = logging.getLogger("cli") class CLITuiRuntimeMixin: """Classic-TUI run loop phases: input processing, after-turn bookkeeping, startup banner/prewarm/maintenance, prompt_toolkit application build, signal handlers and shutdown.""" def _tui_process_loop(self): """REPL worker thread: drain ``_pending_input``, run idle housekeeping, dispatch each input.""" held = None while not self._should_exit: try: if held is not None: held, user_input = self._tui_hold_step(held) if user_input is not None: self._tui_process_one_input(user_input) continue try: user_input = self._pending_input.get(timeout=0.1) except queue.Empty: if not self._agent_running: self._tui_idle_tick() continue from hermes_cli.gateway_queue import OwnQueueHold held = OwnQueueHold.begin(self, user_input) if held is None: self._tui_process_one_input(user_input) except Exception as e: if isinstance(e, OSError) and e.errno == errno.EIO: self._mark_terminal_io_broken("process_loop") logger.warning("process_loop EIO — freezing UI paints (#81521): %s", e) continue logger.warning("process_loop unhandled error (msg may be lost): %s", e) def _tui_hold_step(self, hold): """One step of holding a local input behind this viewer's own ``/queue`` admission (#134889 N5): ``(hold or None, input to run now or None)``. A turn the owner runs now goes first; the held input runs once that admission was taken, left the queue, or the bounded wait ran out.""" with suppress(Exception): self._drain_process_notifications("cli-idle") # adopts the admission on its message.start item = hold.take_gateway_turn(self._pending_input) if item is not None: return hold, item if hold.done(): return None, hold.item time.sleep(0.05) return hold, None def _tui_idle_tick(self): """Idle housekeeping between inputs (agent not running).""" self._check_config_mcp_changes() # auto-reload MCP on mcp_servers change # Termios drift heal first: a drifted tty makes the CLI look dead while the loop is healthy. for step in ( self._check_termios_drift, lambda: self._drain_process_notifications("cli-idle"), self._maybe_fire_loop_tick, self._maybe_resume_parked_goal, ): with suppress(Exception): step() def _tui_dock_drop_dequeued(self): """Re-read the live-work dock once an input has left ``_pending_input``. The spinner thread polls the dock once a second, so without this the prompt a turn just took (a seeded ``-q`` prompt above all) stays listed as queued while the turn starts. If the reply prints in that window the dock shrinks only after it, leaving blank rows under the prompt. The turn's own invalidate repaints the refreshed rows. """ monitor = getattr(self, "_subagent_monitor", None) if monitor is not None: with suppress(Exception): monitor.refresh() def _tui_process_one_input(self, user_input): """Route one submitted input: file drop, /resume pick, ! shell, slash command, or a chat turn.""" from cli import _DIM, _PASTE_REF_RE, _RST, _cprint, _detect_file_drop, _looks_like_slash_command, _strip_leaked_bracketed_paste_wrappers, _strip_leaked_terminal_responses_with_meta from tools.process_registry_notifications import TimelineNotification self._tui_dock_drop_dequeued() user_input, is_voice_input, is_seeded_query = self._tui_unwrap_input(user_input) if not user_input: return notification_preview = user_input if isinstance(user_input, TimelineNotification) else None self._status_bar_suppressed_after_resize = False # input ends post-resize suppression submit_images = [] if isinstance(user_input, tuple): user_input, submit_images = user_input if isinstance(user_input, str): user_input = _strip_leaked_bracketed_paste_wrappers(user_input) user_input, _had_mouse_reports = _strip_leaked_terminal_responses_with_meta(user_input) if _had_mouse_reports: self._recover_terminal_input_modes(reason="mouse reports leaked into submitted input") # A typed bare stop phrase ends an active voice chat (transcripts are checked earlier). if not is_voice_input and self._typed_voice_stop(user_input): return # File drops are detected before any dispatch; seeded -q prompts are literal text. _file_drop = _detect_file_drop(user_input) if isinstance(user_input, str) and not is_seeded_query else None if _file_drop: _drop_path = _file_drop["path"] _remainder = _file_drop["remainder"] if _file_drop["is_image"]: submit_images.append(_drop_path) user_input = _remainder or f"[User attached image: {_drop_path.name}]" _cprint(" " + t("cli.tui.auto_attached_image", name=_drop_path.name)) else: _cprint(" " + t("cli.tui.detected_file", name=_drop_path.name)) user_input = f"[User attached file: {_drop_path}]" + (f"\n{_remainder}" if _remainder else "") elif isinstance(user_input, str): # A bare number right after a bare `/resume` selects that session (never sent to the agent). if self._pending_resume_sessions and self._consume_pending_resume_selection(user_input): return if not is_seeded_query: if self.handle_bang_shell(user_input): return if _looks_like_slash_command(user_input): user_input = self._tui_run_slash_input(user_input) if user_input is None: return if isinstance(user_input, str) and _PASTE_REF_RE.search(user_input): user_input = self._expand_paste_references(user_input) _cprint("") self._print_user_message_preview(notification_preview or user_input) if submit_images: n = len(submit_images) _attached = t("cli.tui.images_attached_one" if n == 1 else "cli.tui.images_attached_other", count=n) _cprint(f" {_DIM}{_attached}{_RST}") self._agent_running = self._interactive_turn = True self._pet_turn_error = self._pet_reasoning = False self._turn_summary_begin() self._app.invalidate() try: self.chat(notification_preview or user_input, images=submit_images or None, voice_input=is_voice_input) finally: self._tui_after_turn() def _tui_run_slash_input(self, user_input: str): """Dispatch a slash command. Returns the pending agent seed to run as a chat turn, else None.""" from cli import _cprint _cprint(f"\n⚙️ {user_input}") try: if not self.process_command(user_input): self._should_exit = True if self._app.is_running: self._app.exit() except KeyboardInterrupt: # Ctrl+C during a slow slash command returns to the prompt instead of exiting. _cprint(f"\n[dim]{t('cli.tui.command_interrupted')}[/dim]") return None _seed, self._pending_agent_seed = self._pending_agent_seed, None return _seed or None def _tui_after_turn(self): """Post-turn bookkeeping after chat() returns (normal, error, or interrupt).""" from cli import _DIM, _RST, _cprint self._agent_running = self._pet_reasoning = False self._spinner_text = self._last_scrollback_tool = "" self._tool_start_time = 0.0 self._pending_tool_info.clear() self.__dict__.pop("_running_tool_labels", None) self._pet_react_turn_end() self._turn_summary_emit() self._interactive_turn = False self._app.invalidate() # After an interrupt the renderer may have drifted (leaked CPR text, VT100 parser # stalled mid-escape): drain stray bytes and force a clean redraw. if self._last_turn_interrupted: self._recover_terminal_after_interrupt() # Re-queue any messages that arrived in _interrupt_queue while the agent was running and were never # claimed by the explicit interrupt path. See _drain_interrupt_queue_to_pending_input for the full # rationale. Regression of #17666 / #18760 — the drain block from the original PR #17939 was # deferred as "worth its own review" and never re-landed (#20271). self._drain_interrupt_queue_to_pending_input() # /goal continuation (queued user input still preempts), then /loop tick completion. for hook, what in ( (self._maybe_continue_goal_after_turn, "goal continuation"), (self._maybe_complete_loop_tick_after_turn, "loop completion"), ): try: hook() except Exception as _exc: logging.debug("%s hook failed: %s", what, _exc) # Continuous voice: restart recording off-thread (beep + recorder start would block process_loop). if self._voice_mode and self._voice_continuous and not self._voice_recording: def _restart_recording(): try: if self._voice_tts: self._voice_tts_done.wait(timeout=60) time.sleep(0.3) # A barge-in capture already owns the mic and submits the interruption itself. if self._voice_barge_capture.is_set(): return self._voice_start_recording() self._app.invalidate() except Exception as e: _cprint(f"{_DIM}{t('cli.voice.autorestart_failed', error=e)}{_RST}") threading.Thread(target=_restart_recording, daemon=True).start() with suppress(Exception): self._drain_process_notifications("cli-post-turn") def _tui_signal_handler(self, signum, frame): """SIGHUP/SIGTERM -> graceful shutdown. The agent is hard-interrupted first so its daemon thread can kill the tool's setsid subprocess group before the main thread unwinds (else an orphan child). ``logger.debug`` is guarded: logging is not reentrant-safe and a shutdown race can raise ``KeyError`` inside the handler, bypassing prompt_toolkit's unwind. """ from cli import _arm_exit_watchdog_on_shutdown_signal, _interrupt_agent_for_signal with suppress(Exception): logger.debug("Received signal %s, triggering graceful shutdown", signum) # Arm the backstop IMMEDIATELY: if the unwind wedges, _run_cleanup never arms its own. # Shutdown intent is now unambiguous — arm the exit backstop IMMEDIATELY, before the graceful unwind # below. If any step of that unwind wedges (main thread parked in a syscall, prompt_toolkit teardown # never returning), _run_cleanup never runs and would never arm its own watchdog — leaving a "dead" # CLI alive for minutes (#65998 class). # Arm the exit backstop now that shutdown intent is unambiguous — covers wedges in the unwind below # that would otherwise leave the process alive with no watchdog (#65998 class). _arm_exit_watchdog_on_shutdown_signal() if self._agent_running: _interrupt_agent_for_signal(self.agent, signum) # Prefer app.exit() over raising KeyboardInterrupt: a KBI from a signal handler # lands in a pt Task ("Unhandled exception in event loop" + "Press ENTER to # continue..."); call_soon_threadsafe lets the loop unwind normally. try: from prompt_toolkit.application.current import get_app_or_none _app = get_app_or_none() _loop = getattr(_app, "loop", None) if _loop is not None: _loop.call_soon_threadsafe(_app.exit) return # clean unwind — no traceback, no ENTER pause except Exception: pass raise KeyboardInterrupt() # fallback for non-prompt_toolkit contexts def _show_startup_browser_notice(self): """Main's ``_show_browser_backend_notice``; on a gateway session the gateway computed the notice (and its once-per-24h stamp) with the session's tools, so it is printed from the snapshot.""" inventory = getattr(getattr(self, "_session_db", None), "banner_inventory", lambda: None)() if inventory is None or "browser_backend_notice" not in inventory: self._show_browser_backend_notice() return notice = inventory.get("browser_backend_notice") if notice: from gateway.warning_notifications import render_notification render_notification(lambda: self._console_print(f"[yellow]⚠ {notice}[/yellow]"), platform="cli") def _tui_print_startup(self): """Startup output: light-mode probe, banner, advisories, resume/welcome lines, tips.""" from cli import _accent_hex, _detect_light_mode with suppress(Exception): # light-mode probe before pt grabs the tty (cached) _detect_light_mode() # Scroll the cursor to the last row so banner, responses and prompt pin to the bottom. with suppress(Exception): _term_lines = shutil.get_terminal_size().lines if _term_lines > 2: print("\n" * (_term_lines - 1), end="", flush=True) from hermes_cli.gateway_client import GatewayClientError try: self._ensure_gateway_session() # the banner shows the gateway session's id and inventory except GatewayClientError as exc: # A refused open (a changed launch flag on --resume) is a startup error, not a traceback. print(f"Error: {exc}", file=sys.stderr) raise SystemExit(2) from None self.show_banner() self._show_security_advisories() self._show_startup_browser_notice() # First-run: an unconfigured install routes into provider onboarding instead of # a chat that spins ~30s and fails with a provider-specific error. TTY only. A # configured profile whose credential is benched or signed out gets the reason instead. try: self._maybe_offer_first_run_setup() except Exception: logger.debug("first-run setup offer failed", exc_info=True) # A persisted Docker sandbox on the previous default image is kept until the user says # so; this TTY is where they can. Either answer pins an image, so it is asked once. if sys.stdin.isatty(): try: from hermes_cli.sandbox_image_switch import offer_interactive from cli import _cprint offer_interactive(cprint=_cprint) except Exception: logger.debug("sandbox image switch offer failed", exc_info=True) # Low disk + auto-prune never decided: ask once (either answer ends it for good). try: from hermes_cli.prune_advisor import offer_interactive as _offer_auto_prune from cli import _cprint _offer_auto_prune(cprint=_cprint) except Exception: logger.debug("auto-prune offer failed", exc_info=True) if self._resumed and self._preload_resumed_session(): self._display_resumed_history() _welcome_skin = None # stays None when the skin engine failed _welcome_text = t("cli.tui.welcome") _welcome_color = "#FFF8DC" try: from hermes_cli.skin_engine import get_active_skin _welcome_skin = get_active_skin() _welcome_text = _welcome_skin.get_branding("welcome", _welcome_text) _welcome_color = _welcome_skin.get_color("banner_text", _welcome_color) except Exception: pass self._console_print(f"[{_welcome_color}]{_welcome_text}[/]") self._tui_startup_prewarm_and_warnings(_welcome_skin) self._print_random_tip() self._tui_startup_background_maintenance() # Before the background preload is folded in (at agent init), show the REQUESTED names. _skills_for_line = self.preloaded_skills or list(self._preload_skills_requested or []) if _skills_for_line and not self._startup_skills_line_shown: self._console_print( f"[bold {_accent_hex()}]{t('cli.tui.activated_skills_label')}[/] {', '.join(_skills_for_line)}") self._startup_skills_line_shown = True self._console_print() def _tui_startup_prewarm_and_warnings(self, _welcome_skin): """Idle-window prewarms (picker cache, agent runtime imports) plus the redaction-off and OpenClaw-residue banners.""" # Warm the /model picker cache off-thread (else its first open blocks ~1-2s). with suppress(Exception): from hermes_cli.model_switch_providers import prewarm_picker_cache_async prewarm_picker_cache_async() # Pre-import the agent runtime (~1.5s: run_agent + OpenAI SDK) off-thread; the import # lock makes an early submit block on the remaining work rather than redo it. # Skipped when Termux defers agent startup on purpose. if os.environ.get("HERMES_DEFER_AGENT_STARTUP") != "1": def _prewarm_agent_runtime() -> None: try: import run_agent import openai except Exception: logger.debug("agent runtime pre-import failed", exc_info=True) threading.Thread(target=_prewarm_agent_runtime, name="agent-runtime-prewarm", daemon=True).start() # Redaction is ON by default; be loud when the operator turned it off. with suppress(Exception): # The redactor snapshots its state at import time so any toggle now won't affect the running # process — we just want the operator to see that they're running without the safety net. See # #17691. _redact_raw = os.getenv("HERMES_REDACT_SECRETS", "true") if _redact_raw.lower() not in {"1", "true", "yes", "on"}: self._console_print( f"[bold red]{t('cli.tui.redaction_disabled_title')}[/] " + t("cli.tui.redaction_disabled_body", value=_escape(_redact_raw), setting="[cyan]security.redact_secrets: true[/]") ) # One-time banner when ~/.openclaw/ is left over from a migration. try: from agent.onboarding import ( OPENCLAW_RESIDUE_FLAG, detect_openclaw_residue, is_seen, mark_seen, openclaw_residue_hint_cli, ) if not is_seen(self.config, OPENCLAW_RESIDUE_FLAG) and detect_openclaw_residue(): try: _resid_color = _welcome_skin.get_color("banner_dim", "#B8860B") except Exception: _resid_color = "#B8860B" self._console_print(f"[{_resid_color}]{openclaw_residue_hint_cli()}[/]") try: from hermes_cli.config import get_config_path as _get_cfg_path_resid mark_seen(_get_cfg_path_resid(), OPENCLAW_RESIDUE_FLAG) except Exception: pass # banner fires again next session except Exception: pass def _tui_startup_background_maintenance(self): """Best-effort startup passes: curator skill maintenance, personal + org skill sync. Off the main thread: the curator's deterministic pass snapshots and prunes the whole skills tree (a due weekly pass held the prompt for 6 minutes on a large library), and the sync pulls can hit the network. The REPL must never wait on housekeeping.""" threading.Thread(target=self._run_startup_maintenance, name="startup-maintenance", daemon=True).start() def _run_startup_maintenance(self): with suppress(Exception): from agent.curator import maybe_run_curator maybe_run_curator( idle_for_seconds=float("inf"), # CLI startup = fully idle on_summary=lambda msg: self._console_print(f"[dim #6b7684]💾 {msg}[/]"), ) def _tui_build_application(self, layout, kb, style): """Construct the prompt_toolkit Application for the REPL.""" from cli import CLI_CONFIG, EditingMode, _STEADY_CURSOR, _select_classic_cli_pt_output _cpr_disabled_output = _select_classic_cli_pt_output(sys.stdout) # Kitty placeholders encode the image id in exact foreground RGB, so the whole app # runs 24-bit there. ColorDepth is imported lazily for tests that stub prompt_toolkit. extra_kw = {} if pet_render.supports_kitty_placeholders(): from prompt_toolkit.output import ColorDepth extra_kw["color_depth"] = ColorDepth.DEPTH_24_BIT if _cpr_disabled_output is not None: extra_kw["output"] = _cpr_disabled_output if _STEADY_CURSOR is not None: extra_kw["cursor"] = _STEADY_CURSOR if EditingMode is not None: # Vi editing mode when display.vim_mode is on. # EMACS is prompt_toolkit's own default, so non-opted-in behaviour is unchanged. extra_kw["editing_mode"] = EditingMode.VI if self._vim_mode else EditingMode.EMACS return Application( layout=layout, key_bindings=kb, style=style, full_screen=False, mouse_support=False, # 0 (default) avoids fighting terminal auto-scroll in non-fullscreen mode. refresh_interval=float(CLI_CONFIG.get("display", {}).get("cli_refresh_interval", 0)), # Erase the bottom chrome on exit instead of freezing a copy into scrollback. # Without this, prompt_toolkit's render_as_done teardown repaints the chrome one last time and # leaves it stranded above the exit summary — so a dead status bar + empty prompt sit between # the conversation transcript and the "Resume this session" block, and stack with the next # session's UI on resume (#38252). The actual conversation transcript is printed through # patch_stdout into normal scrollback and is unaffected; only the managed chrome is erased. # Applies to every exit path (/exit, /quit, EOF, Ctrl+C). erase_when_done=True, **extra_kw, ) def _tui_install_signal_handlers(self): """SIGTERM/SIGHUP -> graceful shutdown; Windows absorbs SIGINT (see body).""" try: import signal as _signal _signal.signal(_signal.SIGTERM, self._tui_signal_handler) if hasattr(_signal, 'SIGHUP'): _signal.signal(_signal.SIGHUP, self._tui_signal_handler) # Windows: absorb SIGINT. Win32 delivers spurious CTRL_C_EVENT when children spawn # from background threads, which would unwind app.run() mid-turn. Real Ctrl+C is # bound by prompt_toolkit. Never call agent.interrupt() here (fake user message). if sys.platform == "win32": _signal.signal(_signal.SIGINT, lambda signum, frame: None) except Exception: pass # restricted environments def _tui_stdin_usable(self) -> bool: """Validate fd 0 before prompt_toolkit starts; on macOS fall back to a select() loop when kqueue can't watch it (uv-managed Python).""" try: os.fstat(0) except OSError: print(t("cli.tui.stdin_unavailable")) return False if sys.platform == "darwin": import selectors as _selectors try: if hasattr(_selectors, "KqueueSelector"): _kq = _selectors.KqueueSelector() try: _kq.register(0, _selectors.EVENT_READ) _kq.unregister(0) finally: _kq.close() except (OSError, ValueError, KeyError): import asyncio as _aio_probe class _SelectEventLoopPolicy(_aio_probe.DefaultEventLoopPolicy): def new_event_loop(self): return _aio_probe.SelectorEventLoop(_selectors.SelectSelector()) _aio_probe.set_event_loop_policy(_SelectEventLoopPolicy()) return True def _closing_session_id(self): """The session quit closes: the agent's, or (gateway) the session opened at the banner before any agent was built. Main created its row lazily at the first turn, so with no agent it had none.""" if self.agent: return self.agent.session_id return self.session_id if getattr(self, "_gateway_session_open", False) else None def _tui_shutdown(self): """Teardown after the app exits: interrupt agent, stop voice/pet, persist + close session, cleanup, exit summary.""" from cli import _DIM, _RST, _cprint, _invoke_interrupted_session_end, _run_cleanup, set_approval_callback, set_secret_capture_callback, set_sudo_password_callback self._should_exit = True self._pet_stop_anim() # Without this line the terminal sits silent through the whole cleanup window. with suppress(Exception): print(f"{_DIM}{t('cli.tui.shutting_down')}{_RST}", flush=True) from hermes_cli.cli_shutdown import _gateway_owns_session_end if self.agent and self._agent_running and not ( self._detach_on_exit and _gateway_owns_session_end(self.agent)): # A gateway turn left by Ctrl+D is not interrupted: ``end_session`` below detaches this # viewer and the owner closes the row once the turn settles (gateway/session_viewer_close). with suppress(Exception): request_hard_interrupt(self.agent) if self._voice_recorder: with suppress(Exception): self._voice_recorder.shutdown() self._voice_recorder = None with suppress(Exception): from tools.voice_mode import cleanup_temp_recordings cleanup_temp_recordings() from agent.vault_backends.unlock import (lock as _vault_lock, set_code_prompt_callback, set_save_login_prompt_callback, set_unlock_prompt_callback) for _unset in (set_sudo_password_callback, set_approval_callback, set_secret_capture_callback, set_unlock_prompt_callback, set_save_login_prompt_callback, set_code_prompt_callback): _unset(None) _vault_lock() # session tokens for external password managers die with the session # On SIGHUP/SIGTERM the agent thread may be reaped before its own persistence runs. self._persist_active_session_before_close() _closing_sid = self._closing_session_id() if self._session_db and _closing_sid: try: self._session_db.end_session(_closing_sid, "cli_close") except (Exception, KeyboardInterrupt) as e: logger.debug("Could not close session in DB: %s", e) if not self._delete_session_on_exit: # Drop the empty row of a start-and-quit session so /resume stays clean. try: self._discard_session_if_empty(_closing_sid) except (Exception, KeyboardInterrupt) as e: logger.debug("Could not prune empty session: %s", e) else: # /exit --delete: remove transcripts + SQLite history. try: _sid = _closing_sid if self._session_db.delete_session(_sid, sessions_dir=get_hermes_home() / "sessions"): _cprint(f" {_DIM}{t('cli.tui.session_deleted', session_id=_escape(_sid))}{_RST}") else: _cprint(f" {_DIM}{t('cli.tui.session_delete_missing', session_id=_escape(_sid))}{_RST}") except (Exception, KeyboardInterrupt) as e: logger.debug("Could not delete session on exit: %s", e) # run_conversation() fires on_session_end on normal completion; only fire here mid-turn. A gateway # turn fires it in the gateway when the interrupted turn finalizes. if self.agent and self._agent_running and not _gateway_owns_session_end(self.agent): _invoke_interrupted_session_end(self.agent, self.agent.session_id, "shutdown") # Ask AND apply before _run_cleanup arms the exit watchdog, so a large archive is never cut # short; a signal-driven exit never asks. import cli as _cli_mod if not _cli_mod._signal_watchdog_armed and _cli_mod._active_worktree: from hermes_cli.worktree_exit import apply_exit_choice, ask_exit_choice with suppress(Exception, KeyboardInterrupt): ask_exit_choice(_cli_mod._active_worktree) apply_exit_choice(_cli_mod._active_worktree, _cprint) _run_cleanup() self._print_exit_summary() self._release_active_session()