hermes-agent/hermes_cli/cli_agent_setup_mixin.py

"""Agent construction + session-resume display for ``HermesCLI``: credential resolution,
per-turn agent config, first-use build, resume preload + recap. ``cli.py`` helpers are
imported lazily inside each method (import cycle)."""

from __future__ import annotations

import sys

from rich.markup import escape as _escape

from agent.i18n import t
from utils import base_url_host_matches


def _single_query_clarify_callback(questions: list) -> dict:
    """Headless clarify answer for ``hermes chat -q``.

    A -q turn never builds the prompt_toolkit app, so the interactive clarify modal
    can never be painted or answered — the CLI callback would poll until
    ``agent.clarify_timeout`` while the caller sees a silent hang. Mirror the oneshot
    path and answer immediately instead.

    The oneshot path answers immediately via ``_oneshot_clarify_callback``; single-query turns need the same
    headless behavior (#94943).
    """
    return {"answers": {}, "outcome": "undelivered", "notice": (
        "single-query mode: no user available to answer. Pick the best choices using your own "
        "judgment, or make the most reasonable assumption you can, and continue.")}


def _current_runtime(cli) -> dict:
    """Snapshot the CLI's resolved provider routing as an AIAgent runtime dict.

    getattr guards stay: tests build minimal shells lacking these attributes."""
    return {
        "api_key": cli.api_key,
        "base_url": cli.base_url,
        "provider": cli.provider,
        "requested_provider": getattr(cli, "requested_provider", cli.provider),
        "api_mode": cli.api_mode,
        "command": cli.acp_command,
        "args": list(cli.acp_args or []),
        "credential_pool": getattr(cli, "_credential_pool", None)}


def _route_signature(model, runtime: dict) -> tuple:
    """Hashable identity of (model, routing) used to detect when the agent must be rebuilt."""
    return (
        model, runtime.get("provider"), runtime.get("requested_provider"), runtime.get("base_url"),
        runtime.get("api_mode"), runtime.get("command"), tuple(runtime.get("args") or ()))


def _cooldown_cause(entry) -> str:
    """Why a benched (exhausted) row is cooling down, from what the pool recorded: a rate-limit or
    quota response, a failed token refresh, or another HTTP failure."""
    reason = (entry.last_error_reason or "").lower()
    if entry.last_error_code in (402, 429) or any(k in reason for k in ("rate", "quota", "insufficient")):
        return t("cli.startup.cooldown_cause_rate_limit")
    if entry.last_error_code is None or "refresh" in reason:
        return t("cli.startup.cooldown_cause_token_refresh")
    return t("cli.startup.cooldown_cause_http", code=entry.last_error_code)


def _credential_pool_notice(provider: str) -> tuple:
    """``(cooling, lines)`` on why *provider*'s pool has nothing selectable right now, for the
    startup notice. *cooling* is True when the first line is a live cooldown with its remaining
    time; a dead (quarantined) sign-in adds a line naming the re-login."""
    import time
    from agent.credential_pool import STATUS_DEAD, STATUS_EXHAUSTED, load_pool
    try:
        pool = load_pool(provider)
        if not pool.has_credentials() or pool.has_available():
            return False, []
        next_at = pool.next_available_at()
        entries = pool.entries()
    except Exception:
        return False, []
    lines = []
    if next_at is not None:
        minutes = max(1, int((next_at - time.time() + 59) // 60))
        benched = [e for e in entries if e.last_status == STATUS_EXHAUSTED]
        cause = _cooldown_cause(benched[0]) if benched else t("cli.startup.cooldown_cause_failed_request")
        lines.append(t("cli.startup.credential_cooling_down", provider=provider, cause=cause, minutes=minutes))
    dead = [e for e in entries if e.last_status == STATUS_DEAD]
    if dead:
        reason = dead[0].last_error_message or dead[0].last_error_reason or t("cli.startup.sign_in_lost")
        lines.append(t("cli.startup.sign_in_was_lost", provider=provider, reason=reason))
    return next_at is not None, lines


def _resume_counts(user_count: int, total: int) -> str:
    """``(N user message(s), M total messages)`` suffix of the resumed-session line."""
    key = "cli.resume.counts_one" if user_count == 1 else "cli.resume.counts_other"
    return t(key, count=user_count, total=total)


def _keyless_custom_base(base_url) -> bool:
    """Custom/local endpoints (llama.cpp, ollama, vLLM) often need no auth; only a
    non-OpenRouter base_url qualifies."""
    return bool(
        isinstance(base_url, str)
        and base_url
        and not base_url_host_matches(base_url, "openrouter.ai"))


def _compression_descendant(session_db, session_id):
    """If ``session_id`` is the (empty) head of a compression chain, return the
    descendant that actually holds the messages; else None. Fails open on DB errors."""
    try:
        resolved_id = session_db.resolve_resume_session_id(session_id)
    except Exception:
        return None
    return resolved_id if resolved_id and resolved_id != session_id else None


def _user_display_text(content) -> str:
    """Recap text for a user row; multimodal lists become text parts + ``[image]`` markers."""
    if isinstance(content, list):
        return " ".join(
            part.get("text", "") if part.get("type") == "text" else "[image]"
            for part in content
            if isinstance(part, dict) and part.get("type") in ("text", "image_url"))
    return "" if content is None else str(content)


def _tool_calls_summary(tool_calls) -> str:
    """``[N tool call(s): name, ...]`` with up to 4 distinct names."""
    names = []
    for tc in tool_calls:
        fn = tc.get("function", {})
        name = fn.get("name", "unknown") if isinstance(fn, dict) else "unknown"
        if name not in names:
            names.append(name)
    names_str = ", ".join(names[:4]) + (", ..." if len(names) > 4 else "")
    key = "cli.resume.tool_calls_one" if len(tool_calls) == 1 else "cli.resume.tool_calls_other"
    return t(key, count=len(tool_calls), names=names_str)


# display_kind -> catalog key of the recap event line (resolved at call time); ``hidden`` rows
# are skipped before this lookup.
_RESUME_EVENT_KEYS = {
    "model_switch": "cli.resume.event_model_changed",
    "async_delegation_complete": "cli.resume.event_delegation_completed",
    "process_complete": "cli.resume.event_process_finished",
    "auto_continue": "cli.resume.event_resumed_interrupted"}

def _collect_resume_entries(display_history, disp: dict, clean_assistant):
    """Displayable ``(role, text)`` recap entries from stored history, truncated per the
    ``display.resume_*`` config; system and tool-result rows are skipped. Returns
    ``(entries, index of last assistant entry, its un-truncated text)``.

    Stored history is untrusted for display: text is sanitized so replay can't clear the
    screen, retitle the window or restyle the panel. Pure-reasoning assistant rows with no
    visible output are skipped, as are tool-call-only rows when ``resume_skip_tool_only``.
    """
    from tools.ansi_strip import sanitize_display_text as _sanitize_display_text
    max_user_len = int(disp.get("resume_max_user_chars", 300))
    max_asst_len = int(disp.get("resume_max_assistant_chars", 200))
    max_asst_lines = int(disp.get("resume_max_assistant_lines", 3))
    skip_tool_only = disp.get("resume_skip_tool_only", True)
    entries: list = []
    last_asst_idx = None
    last_asst_full = None
    for msg in display_history:
        role = msg.get("role", "")
        display_kind = msg.get("display_kind")
        content = msg.get("content")
        tool_calls = msg.get("tool_calls") or []
        if display_kind == "hidden":
            continue
        if display_kind in _RESUME_EVENT_KEYS:
            metadata = msg.get("display_metadata") or {}
            label = metadata.get("display_text") if display_kind in ("async_delegation_complete", "process_complete") else None
            entries.append(("event", _sanitize_display_text(label or t(_RESUME_EVENT_KEYS[display_kind]))))
            continue
        if role == "user":
            text = _sanitize_display_text(_user_display_text(content))
            if len(text) > max_user_len:
                text = text[:max_user_len] + "..."
            entries.append(("user", text))
        elif role == "assistant":
            text = clean_assistant("" if content is None else str(content))
            parts, full_parts = [], []
            if text:
                full_parts.append(text)
                lines = text.splitlines()
                if len(lines) > max_asst_lines:
                    text = "\n".join(lines[:max_asst_lines]) + " ..."
                if len(text) > max_asst_len:
                    text = text[:max_asst_len] + "..."
                parts.append(text)
            if tool_calls:
                parts.append(_tool_calls_summary(tool_calls))
                full_parts.append(parts[-1])
            if not text and (skip_tool_only or not tool_calls):
                continue
            entries.append(("assistant", " ".join(parts)))
            last_asst_idx = len(entries) - 1
            last_asst_full = " ".join(full_parts)
    return entries, last_asst_idx, last_asst_full


# (skin key, fallback) for recap panel colors: body text, session label, border, assistant label.
_RESUME_SKIN_COLORS = (
    ("banner_text", "#FFF8DC"), ("session_label", "#DAA520"), ("session_border", "#8B8682"),
    ("ui_ok", "#8FBC8F"))


def _resume_panel_colors() -> tuple:
    """Active-skin colors for ``_RESUME_SKIN_COLORS`` (fallbacks when no skin loads)."""
    try:
        from hermes_cli.skin_engine import get_active_skin
        _skin = get_active_skin()
        return tuple(_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS)
    except Exception:
        return tuple(default for _, default in _RESUME_SKIN_COLORS)


def _retire_agent(cli) -> None:
    """Drop ``cli.agent`` so the next turn rebuilds it, releasing its LLM clients first: the Codex
    app-server child (and MCP descendants) belongs to the instance, so ``self.agent = None`` alone
    orphans it for the CLI process lifetime (#72548). Session tool state is kept (soft release)."""
    agent = cli.agent
    if agent is not None and hasattr(agent, "release_clients"):
        agent.release_clients()
    cli.agent = None


class CLIAgentSetupMixin:
    """Agent construction + session-resume display methods for ``HermesCLI``."""

    def _ensure_runtime_credentials(self) -> bool:
        """Re-resolve provider credentials before agent use so key rotation / token
        refresh are picked up without restarting the CLI. False on auth failure."""
        from cli import ChatConsole, logger
        from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error
        _primary_exc = None
        runtime = None
        _model_at_entry = self.model
        self._credentials_rate_limited = False
        self._credentials_terminal = False
        try:
            # target_model: the ladder's model-keyed rungs (Zen/Go api_mode, Copilot/Nous
            # api_mode) must see the model this CLI will actually send, not config's `default`,
            # or `hermes -m mimo-v2.5 --provider opencode-go` resolves an api_mode/base_url the
            # sent model cannot use (#112600).
            runtime = resolve_runtime_provider(
                requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
                explicit_base_url=self._explicit_base_url, target_model=self.model or None)
        except Exception as exc:
            _primary_exc = exc
        if _primary_exc is not None:
            runtime = self._resolve_fallback_runtime(_primary_exc)
            if runtime is not None:
                _primary_exc = None
        if runtime is None:
            from hermes_cli.auth import AuthError, is_rate_limited_auth_error
            self._credentials_rate_limited = bool(_primary_exc) and is_rate_limited_auth_error(_primary_exc)
            # Only an explicit re-authentication requirement is terminal. Unknown
            # resolver/network failures must not acquire the sticky Kanban block.
            self._credentials_terminal = isinstance(_primary_exc, AuthError) and _primary_exc.relogin_required
            message = format_runtime_provider_error(_primary_exc) if _primary_exc else t("cli.startup.provider_resolution_failed")
            if getattr(self, "tool_progress_mode", "full") == "off":
                print(message, file=sys.stderr)  # quiet/stream-json: stdout is machine-readable
            else:
                ChatConsole().print(f"[bold red]{message}[/]")
            return False
        api_key = runtime.get("api_key")
        base_url = runtime.get("base_url")
        resolved_provider = runtime.get("provider", "openrouter")
        if resolved_provider != "nous":
            # An explicit provider carries inference. The free-tier identity (for connectors) was
            # created by the boot bootstrap before this point, never here; this prints the one-time
            # "free tier is here" notice the first time an identity is seen beside an own key.
            self._maybe_print_free_tier_available_notice()
        resolved_routing = (
            resolved_provider, runtime.get("api_mode", self.api_mode), runtime.get("command"),
            list(runtime.get("args") or []))
        # A callable api_key is a bearer-token provider (Azure Entra ID): the OpenAI SDK
        # invokes it per request, so skip string validation / placeholder substitution.
        if not callable(api_key) and not (isinstance(api_key, str) and api_key):
            if _keyless_custom_base(base_url):
                # Placeholder key so the SDK doesn't reject the keyless local endpoint.
                api_key = "no-key-required"
                logger.debug(
                    "No API key for custom endpoint %s (source=%s), "
                    "using placeholder — local servers typically ignore auth",
                    base_url, runtime.get("source", ""))
            else:
                _prov = (resolved_provider or self.requested_provider or "").strip()
                if _prov and _prov != "auto":
                    print(f"\n{t('cli.startup.no_api_key_for_provider', provider=_prov)}")
                else:
                    print(f"\n{t('cli.startup.no_provider_configured')}")
                print(f"   {t('cli.startup.run_model_or_setup')}")
                return False
        if not isinstance(base_url, str) or not base_url:
            print(f"\n{t('cli.startup.empty_base_url')}")
            return False
        credentials_changed = api_key != self.api_key or base_url != self.base_url
        routing_changed = resolved_routing != (self.provider, self.api_mode, self.acp_command, self.acp_args)
        self.provider, self.api_mode, self.acp_command, self.acp_args = resolved_routing
        self._credential_pool = runtime.get("credential_pool")
        self._provider_source = runtime.get("source")
        self.api_key = api_key
        self.base_url = base_url

        # A custom_provider entry's explicit `model` wins when the CLI model is unset or
        # is just the provider slug/display name (`hermes chat --model <provider-name>`
        # would otherwise send the provider name as the model string -> 400).
        runtime_model = runtime.get("model")
        if runtime_model and isinstance(runtime_model, str) and (
            not self.model or self.model == self.provider or self.model == runtime.get("name")):
            self.model = runtime_model

        # Still empty (e.g. `hermes auth add` without `hermes model`): fall back to the
        # provider's first catalog model so the API doesn't reject an empty model.
        if not self.model and resolved_provider:
            try:
                from hermes_cli.models import get_default_model_for_provider
                _default = get_default_model_for_provider(resolved_provider)
                if _default:
                    self.model = _default
                    logger.info(
                        "No model configured — defaulting to %s for provider %s",
                        _default, resolved_provider)
            except Exception:
                pass

        # Normalize model for the resolved provider (e.g. swap non-Codex models on openai-codex).
        # Fixes #651.
        model_changed = self._normalize_model_for_provider(resolved_provider)

        # Startup resolved reasoning_config for the launch model; whichever path above moved
        # self.model (auth fallback, custom-entry model, provider default, normalization) leaves a
        # per-model contract the lazily built agent would otherwise miss (an always-thinking model
        # 400s on the primary's effort). Same chokepoint as /model, /new and --resume; an explicit
        # --reasoning is the user's intent for this run and outranks the new model's config.
        if self.model != _model_at_entry and getattr(self, "_explicit_reasoning_config", None) is None:
            from hermes_cli.cli_model_switch_mixin import _resolve_cli_reasoning
            _resolve_cli_reasoning(self)
            logger.info("Model moved to %s: reasoning_config resolved: %s", self.model, self.reasoning_config)

        # AIAgent/OpenAI client holds auth at init, so rebuild on key/routing/model change.
        if (credentials_changed or routing_changed or model_changed) and self.agent is not None:
            _retire_agent(self)
            self._active_agent_route_signature = None
        return True

    def _maybe_print_free_tier_available_notice(self) -> None:
        """One-time notice for installs whose inference is carried by an explicit provider: the free
        tier (inference + connectors) now exists. Printed the first time an identity is present, then
        flagged on that identity so it never repeats. Never blocks or raises."""
        from cli import logger
        try:
            from hermes_cli import anon_auth
            if not anon_auth.guest_notice_pending():
                return
            self._console_print(f"[dim]{anon_auth.FREE_TIER_AVAILABLE_NOTICE}[/]")
            anon_auth.mark_guest_notice_shown()
        except Exception as exc:
            logger.debug("free tier availability notice skipped: %s", exc)

    def _resolve_fallback_runtime(self, primary_exc):
        """Primary provider resolution failed: on an AuthError try each fallback entry in
        order and switch the CLI's requested_provider/model to the first that resolves.
        None when the error is not auth-related or no fallback resolves."""
        from cli import _cprint, logger
        from hermes_cli.auth import AuthError, primary_failure_wording
        from hermes_cli.runtime_provider import resolve_runtime_provider
        if not isinstance(primary_exc, AuthError):
            return None
        _fb_chain = self._fallback_model if isinstance(self._fallback_model, list) else []
        for _fb in _fb_chain:
            _fb_provider = (_fb.get("provider") or "").strip().lower()
            _fb_model = (_fb.get("model") or "").strip()
            if not _fb_provider or not _fb_model:
                continue
            try:
                from hermes_cli.fallback_config import resolve_entry_api_key
                # target_model: the fallback entry names the model that will be sent; without it the
                # ladder keys off config `default` (see _ensure_runtime_credentials, #112600).
                _fb_kwargs = {"requested": _fb_provider, "target_model": _fb_model}
                if _fb.get("base_url"):
                    _fb_kwargs["explicit_base_url"] = _fb["base_url"]
                _fb_api_key = resolve_entry_api_key(_fb)
                if _fb_api_key:
                    _fb_kwargs["explicit_api_key"] = _fb_api_key
                runtime = resolve_runtime_provider(**_fb_kwargs)
                _why_log, _why = primary_failure_wording(primary_exc)  # #117482: quota is not auth
                logger.warning(
                    "Primary provider %s (%s). Falling through to fallback: %s/%s",
                    _why_log, primary_exc, _fb_provider, _fb_model)
                from gateway.warning_notifications import render_notification
                render_notification(
                    lambda: _cprint(t("cli.startup.switching_to_fallback", reason=_why, provider=_fb_provider, model=_fb_model)),
                    platform="cli")
                self.requested_provider = _fb_provider
                self.model = _fb_model
                # reasoning_config follows the swap in _ensure_runtime_credentials (the only caller).
                return runtime
            except Exception:
                continue
        return None

    def _runtime_credentials_ready(self) -> bool:
        """Silently probe whether any inference provider can be resolved.

        Never prints or mutates CLI state, so the interactive first-run path can route a
        keyless install into onboarding before the user types into a chat that can't work.

        See #62935.
        """
        return self._probe_runtime_credentials()[0]

    def _probe_runtime_credentials(self) -> tuple:
        """``(ready, error)``: *error* is the exception that stopped resolution — raised, or
        swallowed by the "auto" ladder and stamped on a keyless fallback — ``None`` when a provider
        resolved (usable or merely keyless). Never prints or mutates CLI state."""
        from hermes_cli.runtime_provider import resolve_runtime_provider
        try:
            runtime = resolve_runtime_provider(
                requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
                explicit_base_url=self._explicit_base_url)
        except Exception as exc:
            return False, exc
        if not isinstance(runtime, dict):
            return False, None
        api_key = runtime.get("api_key")
        base_url = runtime.get("base_url")
        if callable(api_key) or (isinstance(api_key, str) and api_key):
            return bool(base_url), None
        return _keyless_custom_base(base_url), runtime.get("auth_error")

    def _maybe_offer_first_run_setup(self) -> None:
        """Interactive startup gate: a blank install goes to the provider wizard; a configured
        profile whose credential is benched or signed out gets the reason instead (#113720)."""
        if not sys.stdin.isatty():
            return
        ready, error = self._probe_runtime_credentials()
        if not ready and not self._explain_unusable_credentials(error):
            self._offer_first_run_setup()

    def _explain_unusable_credentials(self, error) -> bool:
        """A configured profile whose credential is benched, quarantined or signed out is not a
        blank install: print what is wrong (and the remaining cooldown) instead of the first-run
        wizard, whose "nothing is configured" claim sends operators into a second login that can
        rotate a single-use OAuth grant away from the session that was working (#113720).

        True when the failure was explained; False when nothing is configured (the wizard's case).
        """
        from cli import _cprint
        from hermes_cli.auth import format_auth_error
        if error is None or getattr(error, "code", None) == "no_provider_configured":
            return False
        provider = getattr(error, "provider", None) or self.requested_provider
        cooling, lines = _credential_pool_notice(provider) if provider and provider != "auto" else (False, [])
        _cprint("")
        if cooling:
            # A live cooldown is a wait, not a lost login: lead with it and skip the re-auth hint.
            _cprint(f"⚠️  {_escape(lines.pop(0))}")
            _cprint(f"  {_escape(str(error))}")
        else:
            _cprint(f"⚠️  {_escape(format_auth_error(error))}")
        for line in lines:
            _cprint(f"  {_escape(line)}")
        return True

    def _offer_first_run_setup(self) -> bool:
        """Offer the provider picker when no provider is configured at all (interactive
        startup, TTY). Runs the same flow as ``hermes model`` so onboarding has a single
        source of truth. True when a provider was configured."""
        from cli import _cprint, logger
        _cprint("")
        _cprint(t("cli.startup.first_run_no_provider"))
        _cprint(f"  {t('cli.startup.first_run_pick_provider')}")
        try:
            answer = input(f"  {t('cli.startup.first_run_prompt')} ").strip().lower()
        except (KeyboardInterrupt, EOFError):
            print()
            answer = "n"
        if answer in {"n", "no"}:
            _cprint(f"  {t('cli.startup.first_run_skipped')}")
            return False
        try:
            from hermes_cli.main import select_provider_and_model
            from hermes_cli.observability.shared_metrics_setup import provider_setup_surface
            with provider_setup_surface("cli_setup"):
                select_provider_and_model()
        except (KeyboardInterrupt, EOFError, SystemExit):
            print()
            _cprint(f"  {t('cli.startup.first_run_cancelled')}")
            return False
        except Exception as exc:
            logger.debug("first-run provider setup failed: %s", exc)
            _cprint(f"  {t('cli.startup.provider_setup_failed', error=exc)}")
            _cprint(f"  {t('cli.startup.run_model_to_retry')}")
            return False

        # Re-sync CLI state from what the picker persisted so the next turn uses it without a restart.
        try:
            from hermes_cli.config import load_config
            _model_cfg = (load_config().get("model") or {})
            if isinstance(_model_cfg, dict):
                self.requested_provider = (_model_cfg.get("provider") or "").strip() or self.requested_provider
                _new_model = (_model_cfg.get("default") or _model_cfg.get("model") or "").strip()
                self.model = _new_model or self.model
                # The picker's model has its own per-model reasoning contract (see
                # _resolve_cli_reasoning); an explicit --reasoning stays the user's intent.
                if _new_model and getattr(self, "_explicit_reasoning_config", None) is None:
                    from hermes_cli.cli_model_switch_mixin import _resolve_cli_reasoning
                    _resolve_cli_reasoning(self)
        except Exception as exc:
            logger.debug("first-run config re-sync failed: %s", exc)
        # Force credential re-resolution + agent rebuild on next use.
        _retire_agent(self)
        self._active_agent_route_signature = None
        if self._runtime_credentials_ready():
            _cprint(f"  {t('cli.startup.provider_configured')}")
            return True
        _cprint(f"  {t('cli.startup.provider_setup_incomplete')}")
        return False

    def _resolve_turn_agent_config(self, user_message: str) -> dict:
        """Effective model/runtime config for one turn — always the session's primary
        provider. With a static `/fast` tier (fast / ultrafast) attach request_overrides;
        auto/cold tiers are applied per request by agent.fast_mode instead."""
        from agent.fast_mode import STATIC_TIERS
        from hermes_cli.models import resolve_fast_mode_overrides
        runtime = _current_runtime(self)
        route = {"model": self.model, "runtime": runtime, "signature": _route_signature(self.model, runtime)}
        overrides = None
        tier = getattr(self, "service_tier", None)
        if tier in STATIC_TIERS:
            try:
                overrides = resolve_fast_mode_overrides(
                    route["model"], provider=runtime["provider"], base_url=runtime["base_url"], tier=tier)
            except Exception:
                pass
        route["request_overrides"] = overrides
        return route

    def _follow_compression_chain(self, session_meta, announce):
        """If the resumed id is an empty compression-chain head, announce and switch to
        the descendant holding the messages; returns the (possibly refreshed) meta."""
        resolved_id = _compression_descendant(self._session_db, self.session_id)
        if resolved_id:
            announce(resolved_id)
            self.session_id = resolved_id
            session_meta = self._session_db.get_session(self.session_id) or session_meta
        return session_meta

    def _restore_session_state(self, session_meta, *, quiet: bool = False) -> None:
        """Restore cwd / yolo / model from the resumed session's metadata."""
        self._restore_session_cwd(session_meta, quiet=quiet)
        self._restore_session_yolo(session_meta, quiet=quiet)
        self._restore_session_model(session_meta, quiet=quiet)

    def _reopen_session(self) -> None:
        """Clear ended_at so the resumed session is active again (best effort)."""
        try:
            self._session_db.reopen_session(self.session_id)
        except Exception:
            pass

    def _load_resumed_history_late(self) -> bool:
        """Late resume path: validate the session and load its history from the DB when
        _preload_resumed_session() (called from run()) did not already populate it.
        False when the resume must abort (missing session / over the safe-resume limit)."""
        from cli import ChatConsole, _DIM, _RST, _accent_hex, _cprint
        session_meta = self._session_db.get_session(self.session_id)
        # Quiet mode (tool_progress_mode == "off") routes resume status lines to
        # stderr so stdout stays machine-readable for `$(hermes chat -Q --resume ...)`.
        # Without this, the resume banner pollutes captured stdout. See #11793.
        _quiet_mode = getattr(self, "tool_progress_mode", "full") == "off"

        def _say(plain: str, rich: str) -> None:
            if _quiet_mode:
                print(plain, file=sys.stderr)
            else:
                ChatConsole().print(rich)
        if not session_meta:
            hint = t("cli.resume.session_id_hint")
            not_found = t("cli.resume.session_not_found", session_id=self.session_id)
            if _quiet_mode:
                print(not_found, file=sys.stderr)
                print(hint, file=sys.stderr)
            else:
                _cprint(f"\033[1;31m{not_found}{_RST}")
                _cprint(f"{_DIM}{hint}{_RST}")
            return False
        session_meta = self._follow_compression_chain(
            session_meta,
            lambda rid: ChatConsole().print(
                f"[dim]{_escape(t('cli.resume.compressed_into', session_id=self.session_id, descendant=rid))}[/dim]"))
        if getattr(self, "_resume_history_error", None):
            return False
        # Only the TIP session's rows are loaded here (no ancestors), so use the
        # tip-only count — the full-lineage count would over-reject compressed sessions.
        resume_limit_error = self._resume_history_limit_error(tip_only=True)
        if resume_limit_error:
            self._resume_history_error = resume_limit_error
            _say(
                t("cli.resume.cannot_resume", error=resume_limit_error),
                f"[bold red]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {_escape(resume_limit_error)}")
            return False
        restored = self._session_db.get_messages_as_conversation(self.session_id, repair_alternation=True)
        if restored:
            restored = [m for m in restored if m.get("role") != "session_meta"]
            self.conversation_history = restored
            msg_count = len([m for m in restored if m.get("role") == "user"])
            title_part = f" \"{session_meta['title']}\"" if session_meta.get("title") else ""
            counts = _resume_counts(msg_count, len(restored))
            _say(
                f"{t('cli.resume.resumed_session')} {self.session_id}{title_part} {counts}",
                f"[bold {_accent_hex()}]{_escape(t('cli.resume.resumed_session'))}[/] [bold]{_escape(self.session_id)}[/]"
                f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}")
            self._restore_session_state(session_meta, quiet=_quiet_mode)
        else:
            no_messages = t("cli.resume.session_no_messages", session_id=self.session_id)
            _say(no_messages, f"[bold {_accent_hex()}]{_escape(no_messages)}[/]")
        self._reopen_session()
        return True

    def _init_agent(self, *, model_override: str = None, runtime_override: dict = None, request_overrides: dict | None = None) -> bool:
        """Build the agent on first use; when resuming, restore history from SQLite.
        Returns True on success."""
        from cli import ChatConsole, _cprint, _prepare_deferred_agent_startup, logger
        from run_agent import AIAgent
        if self.agent is not None:
            return True

        # Join the background preloaded-skills load (--skills/-s) BEFORE the agent
        # snapshots self.system_prompt below. No-op when nothing was requested.
        self.finalize_preloaded_skills()
        _prepare_deferred_agent_startup()
        self._install_tool_callbacks()
        if not self._ensure_runtime_credentials():
            return False
        from hermes_cli.mcp_startup import ensure_mcp_discovery_before_agent_build
        ensure_mcp_discovery_before_agent_build(
            logger=logger, single_query=getattr(self, "_single_query_mode", False))
        if self._session_db is None:
            try:
                from hermes_state_registry import acquire
                self._session_db = acquire()
            except Exception as e:
                logger.warning("SQLite session store not available — session will NOT be indexed: %s", e)
        if (
            self._resumed and self._session_db and not self.conversation_history
            and not self._load_resumed_history_late()):
            return False
        try:
            runtime = runtime_override or _current_runtime(self)
            effective_model = model_override or self.model
            # -q never builds the prompt_toolkit app, so the clarify modal can't be
            # answered — answer headless instead of polling until clarify_timeout.
            single_query_mode = getattr(self, "_single_query_mode", False)
            clarify_callback = (
                # See #94943.
                _single_query_clarify_callback
                if single_query_mode
                else self._clarify_callback)
            connection_callback = None if single_query_mode else self._connection_callback
            self.agent = AIAgent(
                model=effective_model, api_key=runtime.get("api_key"),
                base_url=runtime.get("base_url"), provider=runtime.get("provider"),
                requested_provider=runtime.get("requested_provider"),
                api_mode=runtime.get("api_mode"), acp_command=runtime.get("command"),
                acp_args=runtime.get("args"), credential_pool=runtime.get("credential_pool"),
                max_iterations=self.max_turns,
                run_budget_seconds=getattr(self, "run_budget_seconds", None),
                enabled_toolsets=self.enabled_toolsets, disabled_toolsets=self.disabled_toolsets,
                verbose_logging=self.verbose, quiet_mode=not self.verbose,
                tool_progress_mode=getattr(self, "tool_progress_mode", "all"),
                ephemeral_system_prompt=self.system_prompt if self.system_prompt else None,
                prefill_messages=self.prefill_messages or None,
                reasoning_config=self.reasoning_config, service_tier=self.service_tier,
                request_overrides=request_overrides, providers_allowed=self._providers_only,
                providers_ignored=self._providers_ignore, providers_order=self._providers_order,
                provider_sort=self._provider_sort,
                provider_require_parameters=self._provider_require_params,
                provider_data_collection=self._provider_data_collection,
                openrouter_min_coding_score=self._openrouter_min_coding_score,
                session_id=self.session_id, platform="cli", session_db=self._session_db,
                clarify_callback=clarify_callback, connection_callback=connection_callback,
                reasoning_callback=self._current_reasoning_callback(),
                fallback_model=self._fallback_model, thinking_callback=self._on_thinking,
                checkpoints_enabled=self.checkpoints_enabled,
                checkpoint_max_snapshots=self.checkpoint_max_snapshots,
                checkpoint_max_total_size_mb=self.checkpoint_max_total_size_mb,
                checkpoint_max_file_size_mb=self.checkpoint_max_file_size_mb,
                pass_session_id=self.pass_session_id, skip_context_files=self.ignore_rules,
                skip_memory=self.ignore_rules, tool_progress_callback=self._on_tool_progress,
                tool_start_callback=self._on_tool_start if self._inline_diffs_enabled else None,
                tool_complete_callback=self._on_tool_complete if self._inline_diffs_enabled else None,
                stream_delta_callback=self._stream_delta if self.streaming_enabled else None,
                tool_gen_callback=self._on_tool_gen_start if self.streaming_enabled else None,
                notice_callback=self._on_notice, notice_clear_callback=self._on_notice_clear,
                reaction_callback=self._on_reaction)
            # Reference for atexit memory-provider shutdown: ``_run_cleanup`` in cli.py
            # reads ``cli._active_agent_ref``, so this MUST write the ``cli`` module's
            # global — a ``global`` statement here would bind this module's namespace.
            # When this code lived in cli.py a bare ``global _active_agent_ref`` worked; after the god-file
            # extraction into this mixin a ``global`` here would bind *this module's* namespace, leaving
            # ``cli._active_agent_ref`` None forever — so memory shutdown never ran on /exit (#49287).
            import cli as _cli
            _cli._active_agent_ref = self.agent
            # Seed the agent's once-per-lifecycle auto_load cache with the bytes the preload
            # thread rendered, so the shared prompt path never re-reads config or skill files.
            _auto_result = getattr(self, "_auto_load_skills_result", None)
            if _auto_result is not None:
                self.agent._auto_load_skills_result = _auto_result
                self.agent._auto_load_skills_resolved = True
            # Route agent status output through prompt_toolkit so ANSI escapes aren't garbled by
            # patch_stdout's StdoutProxy (#2262), holding lines while a response box streams so a
            # subagent/background completion notice never splits the reply mid-paragraph.
            self.agent._print_fn = self._agent_status_print
            # Hydrate credits notices at session OPEN (parity with the TUI) so a depletion
            # warning shows before the first message. Idempotent + fail-open in the helper.
            try:
                from agent.credits_tracker import seed_credits_at_session_start
                seed_credits_at_session_start(self.agent)
            except Exception:
                pass
            self._active_agent_route_signature = _route_signature(effective_model, runtime)

            # Force-create DB row on /title intent, then apply title.
            if self._pending_title and self._session_db:
                try:
                    self.agent._ensure_db_session()
                    if self.agent._session_db_created:
                        self._session_db.set_session_title(self.session_id, self._pending_title)
                        _cprint(f"  {t('cli.resume.title_applied', title=self._pending_title)}")
                        self._pending_title = None
                    # else: row creation failed transiently — keep _pending_title for retry
                except Exception as e:
                    _cprint(f"  {t('cli.resume.title_apply_failed', error=e)}")
                    # Keep _pending_title so it can be retried after row creation succeeds
            return True
        except Exception as e:
            console = ChatConsole()
            from hermes_cli.cli_chat_error_copy import agent_init_failure_message
            console.print(f"[bold red]{_escape(agent_init_failure_message(e))}[/]")
            from hermes_constants import partial_update_hint
            for line in partial_update_hint(e):
                console.print(line)
            return False

    def _resume_history_limit_error(self, tip_only: bool = False):
        """Return a safe-resume error without materializing transcript rows.

        ``tip_only`` matches call sites that load only the tip session's rows — counting
        the full lineage there would over-reject heavily-compressed sessions with a small
        tip. Generic guard failures fail OPEN; only a genuine over-limit result blocks."""
        if not self._session_db:
            return None
        from cli import logger
        from hermes_state import SessionResumeTooLargeError
        try:
            safety_check = getattr(self._session_db, "assert_resume_safe", None)
            if not callable(safety_check):
                return None
            safety_check(self.session_id, **({"tip_only": True} if tip_only else {}))
        except SessionResumeTooLargeError as exc:
            return str(exc)
        except Exception as exc:
            logger.warning(
                "Resume safety check failed for %s (proceeding without guard): %s",
                self.session_id, exc)
        return None

    def _preload_resumed_session(self) -> bool:
        """Load a resumed session's history early (from run(), before the first chat) so
        it can be displayed; ``_init_agent()`` then skips its own DB round-trip. Sets
        ``self.conversation_history`` and prints the status line. True if history loaded."""
        from cli import _accent_hex
        if not self._resumed or not self._session_db:
            return False
        session_meta = self._session_db.get_session(self.session_id)
        if not session_meta:
            self._console_print(f"[bold red]{_escape(t('cli.resume.session_not_found', session_id=self.session_id))}[/]")
            self._console_print(f"[dim]{_escape(t('cli.resume.session_id_hint'))}[/]")
            return False
        session_meta = self._follow_compression_chain(
            session_meta,
            lambda rid: self._console_print(
                f"[dim]{_escape(t('cli.resume.compressed_into', session_id=self.session_id, descendant=rid))}[/]"))
        # A Kanban worker transcript resumes only through a dispatcher-owned run (#68779) —
        # an ordinary CLI resume would be a live writer the board cannot observe. Refuse AND
        # fall back to a fresh session id, so this process never appends to the worker row.
        from hermes_cli.kanban_resume_guard import kanban_resume_refusal
        if (kanban_refusal := kanban_resume_refusal(self._session_db, self.session_id)):
            self._console_print(f"[bold red]{_escape(kanban_refusal)}[/]")
            self._resumed = False
            from hermes_state_ids import new_session_id
            self.session_id = new_session_id(self.session_start)
            return False
        resume_limit_error = self._resume_history_limit_error()
        if resume_limit_error:
            self._resume_history_error = resume_limit_error
            self._console_print(f"[bold red]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {resume_limit_error}")
            return False
        restored, display_history = self._session_db.get_resume_conversations(self.session_id)
        accent_color = _accent_hex()
        if not restored:
            self._console_print(
                f"[{accent_color}]{_escape(t('cli.resume.session_no_messages', session_id=self.session_id))}[/]")
            return False
        restored = [m for m in restored if m.get("role") != "session_meta"]
        self.conversation_history = restored
        self._resume_display_history = [m for m in display_history if m.get("role") != "session_meta"]
        from agent.context_compressor import is_user_originated_turn
        # Count only user-originated turns: legacy compaction handoffs are durable
        # role=user rows without display_kind.
        msg_count = len([m for m in self._resume_display_history if is_user_originated_turn(m)])
        title_part = f' "{session_meta["title"]}"' if session_meta.get("title") else ""
        self._console_print(
            f"[{accent_color}]{_escape(t('cli.resume.resumed_session'))} [bold]{self.session_id}[/bold]"
            f"{title_part} {_resume_counts(msg_count, len(restored))}[/]")
        self._restore_session_state(session_meta)
        self._reopen_session()
        return True

    def _display_resumed_history(self):
        """Render a dim Rich-panel recap of the previous conversation, capped at the last
        ``resume_exchanges`` user/assistant exchanges with a hidden-count indicator."""
        from cli import CLI_CONFIG, _record_output_history_entry, _strip_reasoning_tags, _suspend_output_history
        from tools.ansi_strip import sanitize_display_text as _sanitize_display_text
        display_history = getattr(self, "_resume_display_history", self.conversation_history)
        if not display_history or self.resume_display == "minimal":
            return
        _disp = CLI_CONFIG.get("display", {})
        entries, _last_asst_idx, _last_asst_full = _collect_resume_entries(
            display_history, _disp, lambda t: _sanitize_display_text(_strip_reasoning_tags(t)))
        if not entries:
            return
        skipped = max(0, len(entries) - int(_disp.get("resume_exchanges", 10)) * 2)
        entries = entries[skipped:]
        # Show the last assistant entry in full so the user sees where they left off.
        if _last_asst_idx is not None and _last_asst_full:
            adj_idx = _last_asst_idx - skipped
            if 0 <= adj_idx < len(entries):
                entries[adj_idx] = ("assistant_last", _last_asst_full)
        from rich.panel import Panel
        from rich.text import Text
        _history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
            _resume_panel_colors())

        # role -> (label, label style, body style, continuation indent)
        you_label = f"  {t('cli.resume.label_you')} "
        assistant_label = f"  {t('cli.resume.label_assistant', agent_name='Hermes')} "
        role_styles = {
            "user": (you_label, f"dim bold {_session_label_c}", "dim", " " * len(you_label)),
            "assistant": (assistant_label, f"dim bold {_assistant_label_c}", "dim", " " * len(assistant_label)),
            "assistant_last": (assistant_label, f"bold {_assistant_label_c}", "", " " * len(assistant_label)),  # full, non-dim
        }
        lines = Text()
        if skipped:
            lines.append(f"  {t('cli.resume.earlier_messages', count=skipped)}\n\n", style="dim italic")
        for i, (role, text) in enumerate(entries):
            if role == "event":
                lines.append(f"  ◈ {text}\n", style="dim italic")
            else:
                label, label_style, body_style, indent = role_styles[role]
                lines.append(label, style=label_style)
                first, *rest = text.splitlines() or [""]  # first line inline, rest indented
                lines.append(first + "\n", style=body_style)
                for ml in rest:
                    lines.append(f"{indent}{ml}\n", style=body_style)
            if i < len(entries) - 1:
                lines.append("")  # small gap
        panel = Panel(
            lines, title=f"[dim {_session_label_c}]{_escape(t('cli.resume.panel_title'))}[/]",
            border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c)
        _record_output_history_entry(lambda: self._render_resume_history_panel_lines(panel))
        with _suspend_output_history():
            self._console_print(panel)