"""HermesCLI constructor phases: display options, model/provider routing, turn limits, toolsets, checkpoints, prompt/reasoning, runtime state, session store and UI state. Mixin split out of ``cli.py``; bound onto ``HermesCLI`` via the MRO. """ from __future__ import annotations from contextlib import contextmanager import logging import os import queue import threading from datetime import datetime from hermes_cli.fallback_config import get_fallback_chain from hermes_state_ids import new_session_id from pathlib import Path from rich.console import Console from typing import Any, Dict, List, Optional from utils import base_url_host_matches, base_url_hostname, is_truthy_value # Log-record parity with the origin module. logger = logging.getLogger("cli") # The launch flags that pick the model route: ``/new`` starts on the profile's default route. _ROUTE_FLAGS = ("model", "provider", "api_key", "base_url") def _adopt_gateway_toolsets(cli, store): """The profile default the gateway resolved for the opened session, when this process has no list to show (display only), and main's "Unknown toolsets" warning as the gateway judged it against its own plugins.""" info = (store.current or {}).get("info") or {} if getattr(cli, "enabled_toolsets", None) is None and isinstance(info.get("enabled_toolsets"), list): cli.enabled_toolsets = list(info["enabled_toolsets"]) if info.get("unknown_toolsets"): from agent.i18n import t as _t names = ", ".join(info["unknown_toolsets"]) cli._console_print(f"[bold red]{_t('cli.startup.unknown_toolsets', names=names)}[/]") def _gateway_launch(cli): """This run's explicit launch flags as gateway ``session.create`` policy (main handed them to AIAgent).""" explicit_model = getattr(cli, "_explicit_model_override", False) provider = getattr(cli, "_explicit_provider", None) return {"model": cli.model if explicit_model else None, "provider": provider if explicit_model and provider not in {None, "", "auto"} else None, "api_key": getattr(cli, "_explicit_api_key", None), "base_url": getattr(cli, "_explicit_base_url", None), # Only a typed -t; the gateway resolves the profile default and the coding posture (#138670). "toolsets": getattr(cli, "enabled_toolsets", None) if getattr(cli, "_explicit_toolsets", False) else None, "reasoning": getattr(cli, "_explicit_reasoning", None), "max_turns": getattr(cli, "_explicit_max_turns", None), "checkpoints": getattr(cli, "_explicit_checkpoints", False) or None, "pass_session_id": getattr(cli, "pass_session_id", False) or None, "ignore_rules": getattr(cli, "ignore_rules", False) or None, "run_budget": getattr(cli, "_explicit_run_budget", None), "verbose": getattr(cli, "verbose", False) or None, "skills": _explicit_launch_skills(cli)} def _explicit_launch_skills(cli): """The `-s/--skills` names, which the gateway renders into the session once at creation (the CLI's own prompt never reaches it). ``skills.auto_load`` names are left out: the gateway's agent loads those.""" requested = getattr(cli, "_preload_skills_requested", None) or [] if not requested: return None from agent.skill_commands import resolve_auto_load_skills from cli import CLI_CONFIG auto = set() if getattr(cli, "ignore_rules", False) else set(resolve_auto_load_skills(CLI_CONFIG)) return [name for name in requested if name not in auto] or None def _record_shell_hook_consent(): """Main's first-use shell-hook prompt, on this terminal. The gateway that registers the hooks has no TTY; it reads the allowlist this writes when the session opens (create or resume). ``--accept-hooks`` / ``HERMES_ACCEPT_HOOKS`` / ``hooks_auto_accept`` approve without asking. A remote gateway has its own home and allowlist, and a bypass launch reads no profile hooks, so nothing is asked for either.""" from hermes_cli.gateway_chat import env_launch_flags flags = env_launch_flags() if os.environ.get("HERMES_TUI_GATEWAY_URL", "").strip() or flags.get("safe_mode") or flags.get("ignore_user_config"): return try: from agent.shell_hooks import record_consent_from_config from hermes_cli.config import load_config record_consent_from_config(load_config()) except Exception: # health: allow BLE001 -- consent is best effort; the gateway skips unapproved hooks logger.warning("shell-hook consent check failed", exc_info=True) @contextmanager def lease_handed_over(cli, held): """Run a gateway attach without this CLI's own active-session lease, then re-claim it. A row the gateway has not imported yet (a legacy / in-process session) is imported under the active-session registry lock, which refuses (``runtime_coordination_required``) while ANY live lease holds the id, including this process's own. The block may move ``cli.session_id`` to the id the gateway returned: the lease is re-claimed for that id (main's exclusivity is kept), or for the unchanged id when the block raises.""" if not held: yield return cli._release_active_session() try: yield except BaseException: cli._claim_active_session("cli", stderr=True) raise if not cli._claim_active_session("cli", stderr=True): raise RuntimeError("session_already_owned") class CLIInitMixin: """HermesCLI constructor phases: display options, model/provider routing, turn limits, toolsets, checkpoints, prompt/reasoning, runtime state, session store and UI state.""" def _init_display_options(self, verbose, compact): """Display-related config: compact/tool-progress/focus view, bells, streaming, previews, stream buffers.""" from cli import CLI_CONFIG, _configure_output_history, _int_or self.console = Console() self.config = CLI_CONFIG # A local /reasoning show|hide or /footer saves the per-platform ``cli`` key: read it (T7). from gateway.display_config import local_display_config display = local_display_config(CLI_CONFIG["display"]) self.compact = compact if compact is not None else display.get("compact", False) # tool_progress: "off" | "new" | "all" | "verbose"; YAML 1.1 parses bare `off` as False. _raw_tp = display.get("tool_progress", "all") self.tool_progress_mode = "off" if _raw_tp is False else str(_raw_tp) # focus_view (/focus) is display-only: snaps tool_progress to "off" (stashing the # pre-focus mode for /focus off); never changes what is sent to the model. self._focus_view_enabled = bool(display.get("focus_view", False)) self._focus_saved_tool_progress = self._focus_last_counted_tool = None self._focus_hidden_lines = 0 if self._focus_view_enabled: from hermes_cli.focus_view import FOCUS_TOOL_PROGRESS_MODE, normalize_tool_progress_mode self._focus_saved_tool_progress = normalize_tool_progress_mode(self.tool_progress_mode) self.tool_progress_mode = FOCUS_TOOL_PROGRESS_MODE self.resume_display = display.get("resume_display", "full") # "full" | "minimal" self.bell_on_complete = display.get("bell_on_complete", False) self.bell_on_prompt = display.get("bell_on_prompt", False) # bell when a blocking modal opens self.show_reasoning = display.get("show_reasoning", True) self.reasoning_full = display.get("reasoning_full", False) _configure_output_history( enabled=display.get("persistent_output", True), max_lines=display.get("persistent_output_max_lines", 200), ) from hermes_cli.cli_modern_layout import busy_input_mode_from_config, layout_from_config, set_active self._modern_layout = layout_from_config(display) # display.cli_layout: modern | classic set_active(self._modern_layout) # busy_input_mode: "interrupt" (redirect the run) | "queue" (next turn) | "steer" (inject mid-run). _bim = busy_input_mode_from_config(display) self.busy_input_mode = _bim if _bim in ("queue", "steer") else "interrupt" # verbose ONLY controls global DEBUG logging; tool_progress="verbose" is independent # (coupling them spewed every module's DEBUG logs to the console). self.verbose = bool(verbose) if verbose is not None else False self.streaming_enabled = display.get("streaming", False) self.show_timestamps = display.get("timestamps", False) self.timestamp_format = display.get("timestamp_format", "%H:%M") _frm = str(display.get("final_response_markdown", "strip")).strip().lower() self.final_response_markdown = _frm if _frm in {"render", "strip", "raw"} else "strip" self._inline_diffs_enabled = display.get("inline_diffs", True) self._inline_images_mode = display.get("inline_images", "auto") # Per-turn accounting: CLI-only chrome riding the tool-progress feed. self._turn_summary_enabled = bool(display.get("turn_summary", True)) self._spinner_token_flow_enabled = bool(display.get("spinner_token_flow", True)) self._turn_summary_collector = None self._turn_summary_start = 0.0 self._turn_token_baseline = 0 self._interactive_turn = False # only run()-loop turns; keeps the summary line off -Q _ump = display.get("user_message_preview", {}) _ump = _ump if isinstance(_ump, dict) else {} self.user_message_preview_first_lines = max(1, _int_or(_ump.get("first_lines", 2), 2)) self.user_message_preview_last_lines = max(0, _int_or(_ump.get("last_lines", 2), 2)) # Streaming display state self._stream_buf = "" # partial line buffer self._reasoning_preview_buf = "" # coalesces tiny reasoning chunks self._stream_started = self._stream_box_opened = self._stream_box_live = False self._streamed_text_this_turn = "" self._held_status_lines: list[str] = [] # agent status lines parked while a box streams # Possible markdown-table lines held until the block ends for wcwidth-aware re-padding. self._stream_table_buf: list[str] = [] self._in_stream_table = False self._pending_edit_snapshots = {} self._last_input_mode_recovery = self._last_termios_drift_check = None # None = never; monotonic epoch is arbitrary self._input_mode_recovery_notice_shown = self._termios_drift_notice_shown = False def _init_model_routing(self, model, toolsets, provider, reasoning, api_key, base_url, max_turns, run_budget, checkpoints, pass_session_id, ignore_rules): """Resolve model/provider/base_url, turn limits, toolsets, checkpoints, prompt/personality, reasoning + routing config.""" self._init_model_and_provider(model, provider, api_key, base_url) self._init_turn_limits(max_turns, run_budget) self._init_toolsets(toolsets) self._init_checkpoints_and_rules(checkpoints, pass_session_id, ignore_rules) self._init_prompt_and_reasoning(reasoning) def _init_model_and_provider(self, model, provider, api_key, base_url): """Priority: CLI args > env vars > config file.""" from cli import CLI_CONFIG, _normalize_moa_model, _split_model_config_default # LLM_MODEL/OPENAI_MODEL env vars are deliberately NOT checked (multi-agent setups # would stomp each other through the environment). _model_config = CLI_CONFIG["model"] # A dict-valued default carries its own provider, which must feed requested_provider # instead of being replaced by the merged model.provider (typically "auto"). _config_model, _nested_provider = _split_model_config_default( _model_config.get("default") or _model_config.get("model") or "" ) # resume must not clobber an explicit -m with the session's stored model. self._explicit_model_override = bool(model) self.model = model or _config_model or "" _cfg_provider = _model_config.get("provider") or os.getenv("HERMES_INFERENCE_PROVIDER") _startup_provider_override = _startup_base_url_override = _startup_api_key_override = "" if self.model: from hermes_cli.model_switch import resolve_startup_model_route _startup_route = resolve_startup_model_route( self.model, explicit_provider=provider or "", current_provider=(provider or _nested_provider or _cfg_provider or ""), user_providers=CLI_CONFIG.get("providers"), custom_providers=CLI_CONFIG.get("custom_providers"), ) if _startup_route is not None: self.model = _startup_route.model _startup_provider_override = _startup_route.provider _startup_base_url_override = _startup_route.base_url _startup_api_key_override = _startup_route.api_key # ``moa:`` selects the MoA virtual provider before provider resolution so the # real provider never sees the unknown model; the prefix wins over --provider. # A ``moa:`` model string selects the MoA virtual provider in one shot (parity with # interactive ``/moa`` and the model picker). See #56828. _moa_provider_override, self.model = _normalize_moa_model(self.model) if self.model == "": # auto-detect from a local server _base_url = _model_config.get("base_url") or "" if base_url_hostname(_base_url) in ("localhost", "127.0.0.1"): from hermes_cli.runtime_provider import _auto_detect_local_model self.model = _auto_detect_local_model(_base_url) or self.model # Provider normalisation may silently override the default but must warn for an # explicit choice (a config model equal to the global fallback is NOT explicit). self._model_is_default = not model and not _config_model # --api-key wins; otherwise a URL-bearing startup alias carries its own credential. # See #28660. self._explicit_api_key = api_key or _startup_api_key_override or None self._typed_api_key = api_key or None # a startup resume it cannot bind is refused; an alias's falls back self._explicit_base_url = base_url or _startup_base_url_override or None # Resolved lazily at use-time via _ensure_runtime_credentials(). self.requested_provider = ( _moa_provider_override or provider or _startup_provider_override or _nested_provider or _cfg_provider or "auto" ) # Only a provider the command line chose is gateway launch policy: the config default is the # gateway's to resolve, and a resume compares launch policy against the session's frozen route. self._explicit_provider = _moa_provider_override or provider or _startup_provider_override or None # `--provider ` without `-m` uses that entry's default_model, else the global # default goes to the custom endpoint and the compressor gets the wrong context length. # Explicit `-m` still wins. See #86978. if not model and provider: try: from hermes_cli.runtime_provider import _get_named_custom_provider _named_custom = _get_named_custom_provider(provider) except Exception as exc: logger.warning( "Could not resolve --provider %s default model; keeping global model.default (%s)", provider, exc, ) _named_custom = None _provider_default = str((_named_custom or {}).get("model") or "").strip() if _provider_default: self.model = _provider_default self._model_is_default = False self._provider_source: Optional[str] = None self.provider = self.requested_provider self.api_mode = "chat_completions" self.acp_command: Optional[str] = None self.acp_args: list[str] = [] self.base_url = ( base_url or _startup_base_url_override or _model_config.get("base_url", "") or os.getenv("OPENROUTER_BASE_URL", "") ) or None # Key matches the resolved base_url; re-resolved by _ensure_runtime_credentials(). # This seed can reach /model before the first turn, so openrouter.ai never gets a real OpenAI key here. _keys = [os.getenv("OPENROUTER_API_KEY"), os.getenv("OPENAI_API_KEY")] if not (self.base_url and base_url_host_matches(self.base_url, "openrouter.ai")): _keys = _keys[::-1] else: from hermes_cli.auth import looks_like_openrouter_key _keys[1] = _keys[1] if looks_like_openrouter_key(_keys[1]) else None self.api_key = api_key or _keys[0] or _keys[1] def _init_turn_limits(self, max_turns, run_budget): """max_turns: CLI arg > config > env var > default; run budget: CLI flag > config.""" from cli import CLI_CONFIG # resolve_turn_limit() accepts "none"/"unlimited" (-> sys.maxsize) alongside ints. # KEEP the root-level CLI_CONFIG["max_turns"] fallback: it is never migrated on disk # and other config paths may bypass the load-time fold. from hermes_cli.config import resolve_turn_limit as _resolve_turn_limit self.max_turns = _resolve_turn_limit(next( (v for v in (max_turns, CLI_CONFIG["agent"].get("max_turns"), CLI_CONFIG.get("max_turns")) if v is not None), os.getenv("HERMES_MAX_ITERATIONS"), )) self.run_budget_seconds = run_budget if run_budget is not None else CLI_CONFIG["agent"].get("run_budget_seconds") self._explicit_run_budget = run_budget # `--run-budget`: session.create policy (the gateway runs the turn) # The gateway builds the agent now: the explicit flag (not the resolved value) rides session.create. self._explicit_max_turns = max_turns def _init_toolsets(self, toolsets): from cli import CLI_CONFIG self.enabled_toolsets = toolsets # Only a picked list rides session.create; _build_cli_from_args clears this unless -t was typed (the # gateway resolves the profile default and coding posture itself; a stale explicit name is refused, #138670). self._explicit_toolsets = toolsets is not None from agent.skill_utils import parse_config_string_list self.disabled_toolsets = parse_config_string_list(CLI_CONFIG["agent"].get("disabled_toolsets")) # No local "Unknown toolsets" check: plugin toolsets are only known where plugins load. The gateway # refuses an unknown picked name at session.create and reports a stale profile one with the session # (``_adopt_gateway_toolsets``), so this process never discovers plugins to tell them apart (PLG-7). def _init_checkpoints_and_rules(self, checkpoints, pass_session_id, ignore_rules): from cli import CLI_CONFIG cp_cfg = CLI_CONFIG.get("checkpoints", {}) if isinstance(cp_cfg, bool): cp_cfg = {"enabled": cp_cfg} self.checkpoints_enabled = checkpoints or cp_cfg.get("enabled", False) self._explicit_checkpoints = bool(checkpoints) self.checkpoint_max_snapshots = cp_cfg.get("max_snapshots", 20) self.checkpoint_max_total_size_mb = cp_cfg.get("max_total_size_mb", 500) self.checkpoint_max_file_size_mb = cp_cfg.get("max_file_size_mb", 10) self.pass_session_id = pass_session_id # --ignore-rules: AIAgent skips context files (AGENTS.md/SOUL.md/...) and memory. self.ignore_rules = ignore_rules or is_truthy_value(os.environ.get("HERMES_IGNORE_RULES")) # The env opt-in is creation policy only: a resume checks just a typed --ignore-rules (the thin client's rule). self._explicit_ignore_rules = bool(ignore_rules) def _init_prompt_and_reasoning(self, reasoning): """Ephemeral system prompt/prefill, reasoning + service tier, OpenRouter routing knobs, fallback chain.""" from cli import CLI_CONFIG, _load_prefill_messages, _parse_reasoning_config, _parse_service_tier_config, _resolve_prefill_messages_file # Env var wins, then hermes_cli.personality (single owner of overlay resolution). from hermes_cli.personality import available_personalities, resolve_ephemeral_system_prompt self.system_prompt = os.getenv("HERMES_EPHEMERAL_SYSTEM_PROMPT", "") or resolve_ephemeral_system_prompt(CLI_CONFIG) self.personalities = available_personalities(CLI_CONFIG) self.prefill_messages = _load_prefill_messages(_resolve_prefill_messages_file(CLI_CONFIG)) # Per-model override > global reasoning_effort. # Reasoning config (OpenRouter reasoning effort level) Per-model override > global reasoning_effort # — resolved through the shared chokepoint in hermes_constants (Closes #21256). from hermes_constants import resolve_reasoning_config self.reasoning_config = resolve_reasoning_config(CLI_CONFIG, self.model) self._explicit_reasoning_config = None self._explicit_reasoning = None # --reasoning wins for this run only (never persisted); unparseable -> warn and ignore. if reasoning is not None and str(reasoning).strip(): _cli_reasoning = _parse_reasoning_config(reasoning) if _cli_reasoning is None: logger.warning("Unknown --reasoning '%s', keeping the configured level", reasoning) else: self.reasoning_config = _cli_reasoning self._explicit_reasoning_config = _cli_reasoning self._explicit_reasoning = str(reasoning).strip() self.service_tier = _parse_service_tier_config(CLI_CONFIG["agent"].get("service_tier", "")) pr = CLI_CONFIG.get("provider_routing", {}) or {} self._provider_sort = pr.get("sort") self._providers_only = pr.get("only") self._providers_ignore = pr.get("ignore") self._providers_order = pr.get("order") self._provider_require_params = pr.get("require_parameters", False) self._provider_data_collection = pr.get("data_collection") # OpenRouter Pareto Code router coding-score floor; out-of-range = unset. _raw_score = (CLI_CONFIG.get("openrouter", {}) or {}).get("min_coding_score") self._openrouter_min_coding_score: Optional[float] = None if _raw_score not in {None, ""}: try: _f = float(_raw_score) if 0.0 <= _f <= 1.0: self._openrouter_min_coding_score = _f except (TypeError, ValueError): pass self._fallback_model = get_fallback_chain(CLI_CONFIG) def _init_runtime_state(self, resume): """Session store + all per-run mutable state (queues, overlays, pet/voice/status-bar fields).""" from cli import _hermes_home from hermes_cli.left_core_migration import serve_notices_on_stderr serve_notices_on_stderr() # A signature change across turns (/model, credential rotation) rebuilds the agent. self._active_agent_route_signature = None self.agent: Optional[Any] = None # initialized on first use self._tool_callbacks_installed = False self._app = None # prompt_toolkit Application (set in run()) self.conversation_history: list[dict[str, Any]] = [] self.session_start = datetime.now() # Per-prompt elapsed timer shown in the status bar. self._prompt_start_time: Optional[float] = None self._prompt_duration: float = 0.0 self._last_turn_finished_at: Optional[float] = None self._init_session_store() self._pending_title: Optional[str] = None self._resumed = bool(resume) self.session_id = resume or new_session_id(self.session_start) getattr(self, "_write_terminal_breadcrumb", lambda: None)() self._history_file = _hermes_home / ".hermes_history" self._last_invalidate: float | None = None # throttles UI repaints (None = never; monotonic epoch is arbitrary) self._init_ui_state() def _init_session_store(self): """Open the session store early (so /title works before the first message).""" self._session_db = None self._session_db_unavailable = False try: # Registry handle, not a bare SessionDB(): goals/loops/heartbeat acquire the same # path a moment later from the REPL thread, and a second writer repeats the full # open (the /proc-wide deleted-WAL scan, ~4k readlinks) while the render thread # holds the GIL — that repeat was the post-banner freeze before the first prompt. from hermes_cli.gateway_session_store import session_store self._session_db = session_store() # the gateway is the single writer except Exception as e: # Without a store the transcript is NOT persisted while the chat looks healthy, # so surface it prominently rather than only logging. # #41386: a failed session store means the transcript is NOT persisted to state.db — the live # chat looks healthy but resume later shows a truncated/empty session. A buried log line is not # enough; surface it prominently so the user knows persistence is off for this run and can fix # the store before relying on resume. self._session_db_unavailable = True logger.warning("Failed to initialize SessionDB — session will NOT be indexed for search: %s", e) from hermes_state_user_copy import describe_storage_failure, storage_failure_details from agent.i18n import t failure = describe_storage_failure(e) def _present_store_warning(): try: Console(stderr=True).print(t( "cli.session_store.unavailable_rich", reason=failure.gloss, action=failure.action, details=storage_failure_details(e))) except Exception: print(t("cli.session_store.unavailable_plain", reason=failure.gloss, action=failure.action)) # Same automatic diagnostic the gateway gates for its home channel (run_notifications). from gateway.warning_notifications import render_notification render_notification(_present_store_warning, platform="cli") # No state.db / checkpoint maintenance here: the gateway (the single writer) runs both. def _ensure_gateway_session(self): """Create (or resume) this CLI's gateway session and adopt the id it returns, before the banner and the first turn. Idempotent; a store without the gateway facade (tests) is left alone.""" from hermes_cli.gateway_session_store import GatewaySessionStore store = getattr(self, "_session_db", None) if getattr(self, "_gateway_session_open", False) or not isinstance(store, GatewaySessionStore): return launch = _gateway_launch(self) held = getattr(self, "_resumed", False) and getattr(self, "_active_session_lease", None) is not None _record_shell_hook_consent() with lease_handed_over(self, held): sid = store.open_session(self, launch) self._gateway_session_open = True from hermes_cli import gateway_plugin_commands, tools_config tools_config._gateway_plugin_toolsets = lambda: gateway_plugin_commands.toolsets(self) _adopt_gateway_toolsets(self, store) from hermes_cli.gateway_tools import bind_session bind_session(self, store.connection) # the session's processes, before _init_agent's chat proxy if sid and sid != self.session_id: self.session_id = sid getattr(self, "_write_terminal_breadcrumb", lambda: None)() def _new_session_id(self, now, *, branch_of=None, title=None): """The id main mints for /new (``new_session_id``) and /branch (``mint_session_id``). On a gateway store it is a session the gateway created: /new is a fresh ``session.create`` on the profile's default model (an explicit -m was for the previous session only) with the run's other launch flags, /branch the child of ``session.mutate branch``. Before this CLI opened its gateway session (or off the gateway) it is main's local mint, which ``_ensure_gateway_session`` replaces on first use, as at startup.""" from hermes_cli.gateway_session_store import GatewaySessionStore store = getattr(self, "_session_db", None) if not isinstance(store, GatewaySessionStore) or not getattr(self, "_gateway_session_open", False): from hermes_state_ids import new_session_id return new_session_id(now) if branch_of: child = store.branch(branch_of, title) # Main's with_session_yolo: a live /yolo carries onto the branch. The gateway's child starts # without the parent's toggle, so set it there too (/branch moves the CLI's own flag over). # Approvals run on the gateway, so this write is not toggle_session_yolo's best-effort persist: # a failure raises before the child shows YOLO locally. from tools.approval import enable_session_yolo, is_session_yolo_enabled if is_session_yolo_enabled(branch_of): store.set_session_yolo(child, True) enable_session_yolo(child) return child # The run's other launch flags hold for every session it opens, as main's one AIAgent did across /new. launch = {key: value for key, value in _gateway_launch(self).items() if key not in _ROUTE_FLAGS} sid = store.create(launch, record_cwd=False) self.session_id = sid return sid def _init_ui_state(self): """Per-run mutable UI state; must exist before any chat() call since -q never goes through run().""" from cli import CLI_CONFIG, _status_bar_visible_from_display_config self._pending_input = queue.Queue() self._interrupt_queue = queue.Queue() self._agent_running = self._should_exit = False self._detach_on_exit = False # Ctrl+D mid-turn: leave the gateway turn running (LC-6b) self._last_turn_interrupted = False # /goal never auto-queues on a Ctrl+C'd turn self._terminal_io_broken = False # stdout EIO: freeze UI paints instead of spinning self._delete_session_on_exit = False # /exit --delete # /update: relaunch() runs from run() after prompt_toolkit restored terminal modes. # /exit --delete: when True, the current session's SQLite history and on-disk transcripts are # deleted during shutdown. Set by process_command() when the user runs /exit --delete or /quit # --delete. Ported from google-gemini/gemini-cli#19332. self._pending_relaunch: list[str] | None = None self._last_ctrl_c_time = 0 # Blocking-prompt overlays (clarify / sudo / approval / slash-confirm / model picker). self._clarify_state = self._clarify_multi_base = None self._clarify_freetext = False self._clarify_prefill = "" self._sudo_state = self._modal_input_snapshot = self._approval_state = None self._slash_confirm_state = self._model_picker_state = None self._clarify_deadline = self._sudo_deadline = self._approval_deadline = self._slash_confirm_deadline = 0 self._approval_lock = threading.Lock() try: # composer placeholder chosen once so it stays stable on screen from hermes_cli.tips import get_random_composer_placeholder self._composer_placeholder = get_random_composer_placeholder() except Exception: self._composer_placeholder = "" self._command_palette_state = self._secret_state = None self._pending_resume_sessions = None # armed by a bare `/resume`; the next bare number selects self._pending_agent_seed = None # one-shot seed from a slash handler self._secret_deadline = 0 self._tool_start_time: float = 0.0 self._pending_tool_info: dict = {} # function_name -> [(preview, args)] for stacked scrollback self._spinner_text = self._command_status = "" self._last_scrollback_tool: str = "" # "new" mode dedup self._command_running = self._command_blocks_input = False # Petdex mascot (display.pet): kitty placeholders on kitty/Ghostty, half-blocks elsewhere. self._pet_renderer = self._pet_anim_thread = None self._pet_slug = self._pet_kitty_pending = "" self._pet_enabled = self._pet_anim_running = False self._pet_cols: int = 18 self._pet_scale: float = 0.7 self._pet_frames_cache: dict = {} self._pet_kitty_cache: dict = {} self._pet_kitty_image_id = self._pet_frame_idx = 0 self._pet_lock = threading.Lock() self._pet_cfg_checked = self._pet_event_until = 0.0 self._pet_event: str = "" self._pet_reasoning = self._pet_turn_error = False self._attached_images: list[Path] = [] self._image_counter = 0 # Ctrl+S prompt stash; in-memory only because drafts routinely contain secrets. from hermes_cli.prompt_stash import PromptStash as _PromptStash self._prompt_stash = _PromptStash() self.preloaded_skills: list[str] = [] self._startup_skills_line_shown = False # skills.auto_load rendered in the preload thread; None until joined. Handed to every # agent this CLI builds so the prompt bytes never depend on when the agent was created. self._auto_load_skills_result: Optional[tuple] = None # Background --skills preload, joined by finalize_preloaded_skills before any agent is built. self._preload_skills_thread: Optional[threading.Thread] = None self._preload_skills_result: Optional[tuple] = None self._preload_skills_error: Optional[BaseException] = None self._preload_skills_requested: list = [] self._preload_skills_finalized = False self._active_session_lease = None # Voice mode state (also reinitialized inside run() for interactive TUI). self._voice_lock = threading.Lock() self._voice_mode = self._voice_tts = self._voice_recording = False self._voice_processing = self._voice_continuous = False self._voice_recorder = self._voice_tts_stop = None self._voice_live_text = "" # live STT partial (stt.streaming) shown in the placeholder self._voice_tts_done = threading.Event() self._voice_tts_done.set() self._voice_barge_capture = threading.Event() # barge monitor is capturing the interruption self._voice_last_tts_text = "" # echo guard self._voice_barge_phase = None # "generation" | "playback" self._status_bar_visible = _status_bar_visible_from_display_config(CLI_CONFIG.get("display")) self._battery_visible = bool(CLI_CONFIG["display"].get("battery", False)) # Vi/vim editing mode for the input composer (display.vim_mode, config-only). # Off by default: prompt_toolkit's standard emacs bindings. self._vim_mode = bool(CLI_CONFIG["display"].get("vim_mode", False)) # Hide rules + status bar until the next input after a resize, so SIGWINCH cannot # stamp a fresh status bar over one the terminal just reflowed into scrollback. self._status_bar_suppressed_after_resize = self._resize_recovery_pending = False self._resize_recovery_lock = threading.Lock() self._resize_recovery_timer = self._status_bar_unsuppress_timer = None # latter: debounced un-suppress self._last_resize_width = None # width change (reflow, needs viewport clear) vs rows-only self._background_tasks: dict[str, threading.Thread] = {} self._background_task_counter = 0 # Cache-hit baseline, reset on model switch / compression so the bar shows the current regime. self._cache_hit_baseline_prompt = self._cache_hit_baseline_read = self._cache_hit_baseline_compressions = 0 self._cache_hit_baseline_model: Optional[str] = None