"""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)

        # 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:<preset>`` selects the MoA virtual provider before provider resolution so the
        # real provider never sees the unknown model; the prefix wins over --provider.
        # A ``moa:<preset>`` 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 <custom>` 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
        # 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