"""Session lifecycle for the interactive CLI: new/resume/save, undo/retry rewinds, yolo
persistence, manual compression, and exit summary.

Mixin bound onto ``HermesCLI`` via the MRO. cli.py-internal symbols are imported LAZILY
inside each method (``from cli import ...``) — never at module load time (import cycle).
"""

from __future__ import annotations

import contextlib
import os
import shutil
import sys

from agent.i18n import t
from hermes_constants import get_hermes_home
from hermes_state_ids import new_session_id
from pathlib import Path
from rich.console import Console
from rich.markup import escape as _escape
from typing import Any, Dict, List, Optional


def _user_turn_indices(history: list) -> list[int]:
    """Indices of *real* user turns: excludes ephemeral scaffolding, display_kind timeline
    rows and compaction handoffs — the same predicate as resume turn counting."""
    from agent.context_compressor import user_originated_turn_view
    from agent.session_persistence import _is_ephemeral_scaffolding

    return [
        i for i, m in enumerate(history)
        if not _is_ephemeral_scaffolding(m) and user_originated_turn_view(m) is not None]


def _timestamp_or(value, default):
    """``datetime.fromtimestamp(value)`` or *default* when the value is missing/unparseable."""
    from cli import datetime

    if not value:
        return default
    try:
        return datetime.fromtimestamp(float(value))
    except Exception:
        return default


def _squash(text: str, limit: int = 120) -> str:
    """Collapse whitespace and cap to *limit* cells with an ellipsis."""
    text = " ".join(text.split())
    return text[: limit - 1] + "…" if len(text) > limit else text


def _dim_notice(cli, msg: str, quiet: bool) -> None:
    """Print a dim notice — raw to stderr on quiet (pre-TUI) paths. Module-level so tests can
    call the resume helpers unbound against a minimal stand-in."""
    if quiet:
        print(msg, file=sys.stderr)
    else:
        cli._console_print(f"[dim]{_escape(msg)}[/dim]")


def _reset_model_to_config_default(cli, silent: bool) -> None:
    """/new is a full boundary: re-derive model/provider from config.yaml so a
    session-only ``/model --session`` switch never leaks into the next session.
    Best-effort — an unreachable default must never block /new. Module-level helper (like
    ``_apply_new_session_title``): tests drive ``new_session`` unbound on a SimpleNamespace."""
    from cli import CLI_CONFIG, _cprint, _split_model_config_default, logger
    _model_config = CLI_CONFIG.get("model", {})
    if isinstance(_model_config, dict):
        _raw_default = _model_config.get("default") or _model_config.get("model") or ""
        _config_provider = _model_config.get("provider", "")
    else:
        _raw_default, _config_provider = (_model_config or ""), ""
    _config_model, _ = _split_model_config_default(_raw_default)
    if not _config_model or _config_model == getattr(cli, "model", None):
        return
    try:
        from hermes_cli.model_switch import switch_model as _switch_model

        r = _switch_model(
            raw_input=_config_model,
            current_provider=cli.provider or "",
            current_model=cli.model or "",
            current_base_url=cli.base_url or "",
            current_api_key=cli.api_key or "",
            is_global=False,
            explicit_provider=_config_provider or "")
        if not r.success:
            return
        if cli.agent:
            cli.agent.switch_model(
                new_model=r.new_model, new_provider=r.target_provider, api_key=r.api_key,
                base_url=r.base_url, api_mode=r.api_mode,
                capabilities=getattr(r, "runtime_capabilities", None))
        cli.model = r.new_model
        cli.provider = r.target_provider
        cli.requested_provider = r.target_provider
        cli._explicit_api_key = r.api_key
        cli._explicit_base_url = r.base_url
        if r.api_key:
            cli.api_key = r.api_key
        if r.base_url:
            cli.base_url = r.base_url
        if r.api_mode:
            cli.api_mode = r.api_mode
        if not silent:
            _cprint(f"  {t('cli.session.model_reset_default', model=r.new_model)}")
    except Exception:
        logger.debug("/new model reset to config default failed", exc_info=True)


def _apply_new_session_title(cli, title: str) -> Optional[str]:
    """Sanitize + persist a /new title; returns the stored title or None (untitled)."""
    from cli import _cprint
    from hermes_state import SessionDB
    try:
        sanitized = SessionDB.sanitize_title(title)
    except ValueError as e:
        _cprint(f"  {t('cli.session.title_rejected', error=e)}")
        return None
    if not sanitized:
        _cprint(f"  {t('cli.session.title_empty_untitled')}")
        return None
    try:
        cli._session_db.set_session_title(cli.session_id, sanitized)
    except ValueError as e:
        _cprint(f"  {t('cli.session.title_error_untitled', error=e)}")
        return None
    except Exception:
        return None
    cli._pending_title = None
    cli._status_bar_title_checked_at = 0.0
    return sanitized


class CLISessionMixin:
    """Session lifecycle for the interactive CLI: new/resume/save, undo/retry rewinds, yolo
    persistence, manual compression, and exit summary."""

    def _restore_session_cwd(self, session_meta: dict, *, quiet: bool = False) -> None:
        """Relaunch a resumed session in the directory it was started from.

        Idempotent; called from every resume path. Both ``os.chdir()`` and ``TERMINAL_CWD``
        are retargeted so the process and the terminal/code-exec tools agree (the local
        terminal backend snapshots cwd on first use, after this). No-op when no cwd was
        recorded, the directory is gone (dim warning, never a crash), or we're already there.
        """
        recorded = (session_meta or {}).get("cwd")
        if not recorded:
            return
        recorded = os.path.expanduser(str(recorded))
        try:
            current = os.getcwd()
        except OSError:
            current = None
        if current and os.path.realpath(recorded) == os.path.realpath(current):
            return
        if not os.path.isdir(recorded):
            msg = t("cli.session.cwd_gone", recorded=recorded, current=current or ".")
        else:
            try:
                os.chdir(recorded)
                os.environ["TERMINAL_CWD"] = recorded
                msg = t("cli.session.cwd_restored", path=recorded)
            except OSError as e:
                msg = t("cli.session.cwd_enter_failed", path=recorded, error=e)
        _dim_notice(self, msg, quiet)

    def _restore_session_yolo(self, session_meta: dict, *, quiet: bool = False) -> None:
        """Re-enable YOLO bypass on resume when the session row's ``model_config.yolo_mode``
        says so — the in-memory ``tools.approval._session_yolo`` set starts empty in a fresh
        process. No-op when already active or when the process was launched with ``--yolo``."""
        try:
            from hermes_state import SessionDB
            from tools.approval import (
                _YOLO_MODE_FROZEN, enable_session_yolo, is_session_yolo_enabled)
        except Exception:
            return
        if _YOLO_MODE_FROZEN or not SessionDB.session_yolo_enabled(session_meta):
            return
        session_key = self.session_id or "default"
        if is_session_yolo_enabled(session_key):
            return
        enable_session_yolo(session_key)
        _dim_notice(self, t("cli.session.yolo_restored"), quiet)

    def _render_resume_history_panel_lines(self, panel) -> list[str]:
        """Render the resume panel at the current terminal width for resize replay."""
        from cli import _suspend_output_history
        from io import StringIO

        buf = StringIO()
        console = Console(
            file=buf, force_terminal=True, color_system="truecolor", highlight=False,
            width=shutil.get_terminal_size((80, 24)).columns)
        with _suspend_output_history():
            console.print(panel)
        return buf.getvalue().rstrip("\n").splitlines()

    def _resolve_checkpoint_ref(self, ref: str, checkpoints: list) -> str | None:
        """Resolve a 1-indexed checkpoint number (or pass through a git hash)."""
        try:
            idx = int(ref) - 1
        except ValueError:
            return ref
        if 0 <= idx < len(checkpoints):
            return checkpoints[idx]["hash"]
        print(f"  {t('cli.session.invalid_checkpoint', count=len(checkpoints))}")
        return None

    def _show_status(self):
        """Show compact startup status line."""
        from cli import get_tool_definitions
        # Avoid pulling the full tool registry into the bare Termux prompt path.
        if os.environ.get("HERMES_DEFER_AGENT_STARTUP") == "1":
            tool_status = t("cli.session.status_tools_deferred")
        else:
            tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets,
                                         disabled_toolsets=self.disabled_toolsets, quiet_mode=True)
            tool_status = t("cli.session.status_tools_count", count=len(tools) if tools else 0)

        model_short = self.model.split("/")[-1] if "/" in self.model else self.model
        if len(model_short) > 30:
            model_short = model_short[:27] + "..."
        api_indicator = "[green bold]●[/]" if self.api_key else "[red bold]●[/]"
        try:
            from hermes_cli.skin_engine import get_active_skin
            skin = get_active_skin()
            separator_color = skin.get_color("banner_dim", "#B8860B")
            accent_color = skin.get_color("ui_accent", "#FFBF00")
            label_color = skin.get_color("ui_label", "#DAA520")
        except Exception:
            separator_color, accent_color, label_color = "#B8860B", "#FFBF00", "cyan"
        sep = f" [dim {separator_color}]·[/] "
        toolsets_info = ""
        if self.enabled_toolsets and "all" not in self.enabled_toolsets:
            toolsets_info = f"{sep}[{label_color}]{_escape(t('cli.session.status_toolsets', toolsets=', '.join(self.enabled_toolsets)))}[/]"
        provider_info = f"{sep}[dim]{_escape(t('cli.session.status_provider', provider=self.provider))}[/]"
        if self._provider_source:
            provider_info += f"{sep}[dim]{_escape(t('cli.session.status_auth', source=self._provider_source))}[/]"
        self._console_print(
            f"  {api_indicator} [{accent_color}]{model_short}[/]{sep}"
            f"[bold {label_color}]{tool_status}[/]{toolsets_info}{provider_info}")

    def _show_session_status(self):
        """Show gateway-style status for the current CLI session."""
        from hermes_cli.status_report import build_status_fields, status_lines
        session_meta = {}
        if self._session_db:
            with contextlib.suppress(Exception):
                session_meta = self._session_db.get_session(self.session_id) or {}

        agent = getattr(self, "agent", None)
        fields = build_status_fields(
            self.session_id, agent, session_meta,
            model=getattr(self, "model", None), provider=getattr(self, "provider", None),
            created_fallback=self.session_start, agent_running=bool(getattr(self, "_agent_running", False)),
        )

        reasoning_label = None
        rc = getattr(agent, "reasoning_config", None) or getattr(self, "reasoning_config", None)
        if isinstance(rc, dict):
            if rc.get("enabled") is False:
                reasoning_label = t("cli.shared.label_off")
            elif rc.get("effort"):
                reasoning_label = str(rc.get("effort"))
        show_r = getattr(self, "show_reasoning", None)
        if reasoning_label and show_r is not None:
            state = t("cli.shared.label_on") if show_r else t("cli.shared.label_off")
            reasoning_label += t("cli.session.reasoning_display_suffix", state=state)

        approval_label = None
        try:
            from tools.approval import is_approval_bypass_active_for_session
            from tools.approval_context import _get_approval_mode
            approval_label = _get_approval_mode()
            if is_approval_bypass_active_for_session(getattr(self, "session_key", "") or ""):
                approval_label += t("cli.session.yolo_bypass_suffix")
        except Exception:
            pass

        # Context window usage: reuse the status-bar snapshot (tokens / max / percent).
        ctx_label = None
        try:
            snap = self._get_status_bar_snapshot()
            ctx_max = snap.get("context_length")
            ctx_pct = snap.get("context_percent")
            if ctx_max:
                left = ""
                if isinstance(ctx_pct, (int, float)):
                    left = t("cli.session.ctx_left_prefix", percent=max(0, 100 - int(ctx_pct)))
                ctx_label = t("cli.session.ctx_tokens_used", left=left, used=f"{snap.get('context_tokens') or 0:,}",
                              max=f"{ctx_max:,}")
        except Exception:
            ctx_label = None

        lines = [t("cli.session.status_title"), "", *status_lines(fields, "session_id", "path", "title", "model")]
        try:
            from hermes_cli.anon_auth import free_tier_route

            if free_tier_route():
                lines.append(t("gateway.status.free_tier"))
        except Exception:
            pass
        optional = ((t("cli.session.label_reasoning"), reasoning_label),
                    (t("cli.session.label_approvals"), approval_label),
                    (t("cli.session.label_context"), ctx_label))
        for label, value in optional:
            if value:
                lines.append(t("cli.shared.label_value", label=label, value=value))
        lines.extend(status_lines(fields, "created", "last_activity", "tokens", "agent_running"))
        self._console_print("\n".join(lines), highlight=False, markup=False)

    def _list_recent_sessions(self, limit: int = 10) -> list[dict[str, Any]]:
        """Return recent CLI sessions for in-chat browsing/resume affordances."""
        if not self._session_db:
            return []
        try:
            from hermes_cli.session_listing import query_session_listing
            from hermes_state_sessions import INTERNAL_LISTING_SOURCES

            return query_session_listing(
                self._session_db, source="cli", current_session_id=self.session_id,
                include_all_sources=False, include_unnamed=True, limit=limit,
                exclude_sources=list(INTERNAL_LISTING_SOURCES))
        except Exception:
            return []

    def _show_recent_sessions(self, *, reason: str = "history", limit: int = 10) -> bool:
        """Render recent sessions inline from the active chat TUI.

        Returns True when something was shown, False if no session list was available.
        """
        from cli import _cli_visible_print
        sessions = self._list_recent_sessions(limit=limit)
        if not sessions:
            return False

        from hermes_cli.timefmt import relative_time as _relative_time

        _cli_visible_print()
        if reason == "history":
            _cli_visible_print(t("cli.session.recent_intro"))
        else:
            _cli_visible_print(f"  {t('cli.session.recent_header')}")
        _cli_visible_print()
        _cli_visible_print(
            f"  {'#':<3} {t('cli.session.column_title'):<32} {t('cli.session.column_preview'):<40} "
            f"{t('cli.session.column_last_active'):<13} {t('cli.session.column_id')}")
        _cli_visible_print(f"  {'─' * 3} {'─' * 32} {'─' * 40} {'─' * 13} {'─' * 24}")
        for idx, session in enumerate(sessions, start=1):
            title = session.get("title") or "—"
            preview = (session.get("preview") or "")[:38]
            last_active = _relative_time(session.get("last_active"), session_id=session.get("id"))
            _cli_visible_print(f"  {idx:<3} {title:<32} {preview:<40} {last_active:<13} {session['id']}")
        _cli_visible_print()
        _cli_visible_print(f"  {t('cli.session.resume_usage')}")
        _cli_visible_print(f"  {t('cli.session.resume_example')}")
        _cli_visible_print()
        return True

    def show_history(self):
        """Display conversation history."""
        from cli import _cli_visible_print
        if not self.conversation_history:
            if not self._show_recent_sessions(reason="history"):
                _cli_visible_print(t("cli.session.history_empty"))
            return

        preview_limit = 400
        visible_index = 0
        hidden_tool_messages = 0
        show_ts = bool(getattr(self, "show_timestamps", False))

        def _ts_suffix(message: dict) -> str:
            # Only annotate when the toggle is on AND the turn has a stored unix
            # `timestamp` (SessionDB-restored rows do; live turns may not) — never fabricate.
            ts = message.get("timestamp") if show_ts else None
            if not ts:
                return ""
            try:
                stamp = _timestamp_or(ts, None)
                return f"  [{stamp.strftime(getattr(self, 'timestamp_format', '%H:%M'))}]" if stamp else ""
            except (ValueError, OSError, TypeError):
                return ""

        def flush_tool_summary():
            nonlocal hidden_tool_messages
            if not hidden_tool_messages:
                return
            key = "cli.session.history_tools_hidden_one" if hidden_tool_messages == 1 else "cli.session.history_tools_hidden_other"
            _cli_visible_print(f"\n  {t('cli.session.history_tools_header')}")
            _cli_visible_print(f"    {t(key, count=hidden_tool_messages)}")
            hidden_tool_messages = 0

        rule = "+" + "-" * 50 + "+"
        for line in ("", rule, "|" + t("cli.session.history_title").center(50) + "|", rule):
            _cli_visible_print(line)

        for msg in self.conversation_history:
            role = msg.get("role", "unknown")
            if role == "tool":
                hidden_tool_messages += 1
                continue
            if role not in {"user", "assistant"}:
                continue
            flush_tool_summary()
            visible_index += 1

            content = msg.get("content")
            content_text = "" if content is None else str(content)
            preview = content_text[:preview_limit]
            suffix = "..." if len(content_text) > preview_limit else ""
            if role == "user":
                _cli_visible_print(f"\n  {t('cli.session.history_you_header', index=visible_index)}{_ts_suffix(msg)}")
                _cli_visible_print(f"    {preview}{suffix}")
                continue

            _cli_visible_print(
                f"\n  {t('cli.session.history_assistant_header', agent_name='Hermes', index=visible_index)}{_ts_suffix(msg)}")
            n_calls = len(msg.get("tool_calls") or [])
            if not content_text:
                suffix = ""
                preview = t("cli.session.history_no_text")
                if n_calls:
                    key = "cli.session.history_tool_calls_one" if n_calls == 1 else "cli.session.history_tool_calls_other"
                    preview = t(key, count=n_calls)
            _cli_visible_print(f"    {preview}{suffix}")

        flush_tool_summary()
        _cli_visible_print()

    def _notify_session_boundary(self, event_type: str) -> None:
        """Fire a session-boundary plugin hook (on_session_finalize / on_session_reset).
        Non-blocking; errors swallowed. Safe from shutdown, /new, /reset."""
        with contextlib.suppress(Exception):
            from hermes_cli.lifecycle import finalize_session, invoke_hook

            context = {
                "session_id": self.agent.session_id if self.agent else None,
                "platform": getattr(self, "platform", None) or "cli",
                "reason": "new_session" if event_type == "on_session_reset" else "session_boundary"}
            if event_type == "on_session_finalize":
                finalize_session(**context)
            else:
                invoke_hook(event_type, **context)

    def _discard_session_if_empty(self, session_id: Optional[str]) -> bool:
        """Drop a just-ended session row that never gained content (quit-immediately, /new,
        /clear) so it doesn't clutter ``/resume``. ``SessionDB.delete_session_if_empty`` only
        removes rows with no messages, no title and no children (gemini-cli#27770 port)."""
        from cli import logger
        if not self._session_db or not session_id:
            return False
        # In-memory transcript is authoritative: a real conversation whose DB flush failed
        # or hasn't happened yet must never be pruned.
        if getattr(self, "conversation_history", None):
            return False
        try:
            from hermes_constants import get_hermes_home as _ghh
            return self._session_db.delete_session_if_empty(
                session_id, sessions_dir=_ghh() / "sessions")
        except Exception:
            logger.debug("Could not prune empty session %s", session_id, exc_info=True)
            return False

    def _launch_session_boundary_memory_flush(
        self, history_snapshot: list, *, session_id: Optional[str] = None) -> Optional[list]:
        """Stage old-session memory extraction so /new stays responsive.

        The context-engine ``on_session_end`` is delivered synchronously here: cheap (no LLM)
        and ordering-sensitive — it must land before ``reset_session_state()`` rebinds the
        engine. The memory-provider half (LLM-bound, seconds) is NOT run here: the returned
        snapshot goes to ``MemoryManager.commit_session_boundary_async`` as one end→switch
        task on the serialized worker, so a late ``on_session_end`` can never run after
        ``on_session_switch`` and misattribute the old transcript to the new session.

        Returns the snapshot to queue, or ``None`` when there is nothing to extract.
        """
        from cli import logger
        agent = getattr(self, "agent", None)
        if not agent or not history_snapshot:
            return None
        engine = getattr(agent, "context_compressor", None)
        if engine is not None and hasattr(engine, "on_session_end"):
            try:
                engine.on_session_end(session_id or "", history_snapshot)
            except Exception:
                logger.debug("Context engine on_session_end failed at /new boundary", exc_info=True)
        # No memory manager → new_session() falls back to the inline switch path.
        if getattr(agent, "_memory_manager", None) is None:
            return None
        return history_snapshot

    def new_session(self, silent=False, title=None):
        """Start a fresh session with a new session ID and cleared agent state."""
        from cli import (
            CLI_CONFIG, _parse_service_tier_config,
            _sync_process_session_id, datetime)
        from hermes_cli.cli_model_switch_mixin import _resolve_cli_reasoning
        old_session_id = self.session_id
        _boundary_snapshot = None
        if self.agent:
            if self.conversation_history:
                # Context-engine boundary now; provider extraction is queued below (after
                # rotation) so /new never blocks on the LLM-bound call.
                _boundary_snapshot = self._launch_session_boundary_memory_flush(
                    list(self.conversation_history), session_id=old_session_id)
            self._notify_session_boundary("on_session_finalize")

        if self._session_db and old_session_id:
            # /new can arrive mid-turn before _flush_messages_to_session_db() ran — flush
            # the current turn to the OLD session before rotating or it is silently lost.
            if self.agent:
                with contextlib.suppress(Exception):
                    # Flush any un-persisted messages from the current turn to the old session *before*
                    # rotating.  /new can be called mid-turn when _flush_messages_to_session_db() has not
                    # yet run — without this, messages generated during the current turn are silently lost
                    # on session rotation (#47202).
                    # See #47202.
                    # See #47202.
                    self.agent._flush_messages_to_session_db(
                        self.conversation_history, conversation_history=self.conversation_history)
            with contextlib.suppress(Exception):
                self._session_db.end_session(old_session_id, "new_session")
            self._discard_session_if_empty(old_session_id)

        self.session_start = datetime.now()
        self.session_id = new_session_id(self.session_start)
        # getattr: tests drive new_session unbound against a SimpleNamespace stand-in.
        getattr(self, "_write_terminal_breadcrumb", lambda: None)()
        self.conversation_history = []
        self._pending_title = None
        self._resumed = False
        # An explicit -m/--model was for the previous session only.
        self._explicit_model_override = False
        # Session-scoped overrides (/model --session, /fast, one-turn restores) don't carry over.
        # Re-derive model/provider and service tier from config.yaml so a session-only switch never leaks
        # into the next session (#48055, #23131).
        self._pending_one_turn_model_restore = None
        self.service_tier = _parse_service_tier_config(CLI_CONFIG["agent"].get("service_tier", ""))
        _reset_model_to_config_default(self, silent)
        # After the model reset: the effort belongs to the model the fresh session lands on (a /reasoning
        # session override is dropped, the default model's per-model override is kept).
        _resolve_cli_reasoning(self)
        _sync_process_session_id(self.session_id)

        if self.agent:
            self.agent.session_id = self.session_id
            self.agent.session_start = self.session_start
            self.agent.reasoning_config = self.reasoning_config
            self.agent.reset_session_state()
            if hasattr(self.agent, "_last_flushed_db_idx"):
                self.agent._last_flushed_db_idx = 0
            if hasattr(self.agent, "_todo_store"):
                with contextlib.suppress(Exception):
                    from tools.todo_tool import TodoStore
                    self.agent._todo_store = TodoStore()
            if hasattr(self.agent, "_invalidate_system_prompt"):
                self.agent._invalidate_system_prompt()

            if self._session_db:
                with contextlib.suppress(Exception):
                    self.agent._session_db_created = False
                    self._session_db.create_session(
                        session_id=self.session_id,
                        source=os.environ.get("HERMES_SESSION_SOURCE", "cli"),
                        model=self.model,
                        model_config={
                            "max_iterations": self.max_turns, "reasoning_config": self.reasoning_config,
                        })
                    self.agent._session_db_created = True
                if title:
                    title = _apply_new_session_title(self, title)
            # Tell memory providers the session_id rotated (reset=True flushes per-session
            # state) BEFORE the plugin on_session_reset hook. With old history, end-of-session
            # extraction and this switch are queued as ONE task on the serialized worker —
            # end strictly before switch, without blocking /new. No history → switch inline.
            _mm = getattr(self.agent, "_memory_manager", None)
            with contextlib.suppress(Exception):
                if _mm is not None and _boundary_snapshot:
                    _mm.commit_session_boundary_async(
                        _boundary_snapshot, new_session_id=self.session_id,
                        parent_session_id=old_session_id or "", reason="new_session")
                elif _mm is not None:
                    _mm.on_session_switch(
                        self.session_id, parent_session_id=old_session_id or "",
                        reset=True, reason="new_session")
            self._notify_session_boundary("on_session_reset")

        if not silent:
            if title:
                print(t("cli.session.new_session_titled", title=title))
            else:
                print(t("cli.session.new_session"))

    def _consume_pending_resume_selection(self, text: str) -> bool:
        """Resolve a bare numeric reply following a bare ``/resume`` prompt.

        ``/resume`` (no args) arms ``self._pending_resume_sessions``; the next input gets one
        chance to be a bare session number. The pending state is one-shot — cleared on the
        first input regardless of outcome, so a stray later number is never hijacked.
        Returns True if the input was consumed (caller must not treat it as chat).

        See #34584.
        """
        from cli import _cprint
        pending = self._pending_resume_sessions
        if not pending:
            return False
        self._pending_resume_sessions = None
        if not isinstance(text, str):
            return False
        # Only a pure number selects; "/resume 3", titles etc. fall through.
        if not text.strip().isdigit():
            return False
        index = int(text.strip())
        if not 1 <= index <= len(pending):
            _cprint(f"  {t('cli.session.resume_index_out_of_range', index=index)}")
            _cprint(f"  {t('cli.session.resume_no_args_hint')}")
            return True
        self._handle_resume_command(f"/resume {index}")
        return True

    def save_conversation(self, cmd: str = "/save"):
        """Handle ``/save [json|md|html] [filename] [redact]``.

        A convenience export only — every message is already persisted to the session DB, so
        the live session stays resumable regardless. ``redact`` runs the force-mode secret
        redaction pass before writing.
        """
        from cli import datetime
        from hermes_cli.session_export import (
            SAVE_TRANSCRIPT_FORMATS, SAVE_USAGE, normalize_save_format, render_session_for_save)

        parts = cmd.split()[1:]
        redact = bool(parts) and parts[-1].lower() in ("redact", "--redact")
        if redact:
            parts = parts[:-1]
        if not parts:
            print(SAVE_USAGE)
            return
        try:
            fmt = normalize_save_format(parts[0])
        except ValueError as e:
            print(f"(._.) {e}")
            print(SAVE_USAGE)
            return
        filename = parts[1] if len(parts) > 1 else None

        # Prefer the durable DB row (metadata + tool calls); fall back to in-memory history.
        # getattr: test doubles may not carry _session_db / session_id.
        session_data = None
        _db = getattr(self, "_session_db", None)
        _sid = getattr(self, "session_id", None)
        if _db and _sid:
            try:
                session_data = _db.export_session(_sid, include_compacted=fmt in SAVE_TRANSCRIPT_FORMATS)
            except Exception:
                session_data = None
        if not session_data:
            if not self.conversation_history:
                print(t("cli.session.save_nothing"))
                return
            session_data = {
                "id": self.session_id, "model": self.model,
                "started_at": self.session_start.timestamp(), "messages": self.conversation_history}
        if redact:
            from hermes_cli.session_export_md import redact_session_data

            session_data = redact_session_data(session_data)

        saved_dir = get_hermes_home() / "sessions" / "saved"
        try:
            saved_dir.mkdir(parents=True, exist_ok=True)
        except Exception as e:
            print(t("cli.session.save_dir_failed", path=saved_dir, error=e))
            return
        if filename:
            path = Path(filename).expanduser()
            if not path.is_absolute():
                path = Path.cwd() / path
        else:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            path = saved_dir / f"hermes_conversation_{timestamp}.{fmt}"

        try:
            content = render_session_for_save(session_data, fmt)
            with open(path, "w", encoding="utf-8") as f:
                f.write(content)
            label = {"json": "JSON", "md": t("cli.session.format_markdown"), "html": "HTML"}[fmt]
            print(t("cli.session.save_ok", path=path, label=label))
            # #76354 review F5: the worker thread also rebound the session ContextVar inside its own
            # (copied) context, which the caller never sees — and get_session_env() prefers an already-bound
            # ContextVar over os.environ. Rebind in the CALLER's context so post-compression
            # tools/subprocesses on this thread resolve HERMES_SESSION_ID to the child id after an
            # out-of-place rotation (idempotent when no rotation happened).
            if self.session_id:
                print(f"       {t('cli.session.save_resume_hint', session_id=self.session_id)}")
        except Exception as e:
            print(t("cli.session.save_failed", error=e))

    def _publish_truncated_history(self, truncated: list, *, invalidate_prompt: bool) -> None:
        """Install a rewound history and mirror it onto the agent (flush index reset so the
        next turn re-flushes from the truncated head)."""
        self.conversation_history = truncated
        agent = self.agent
        if agent is None:
            return
        if invalidate_prompt and hasattr(agent, "_invalidate_system_prompt"):
            with contextlib.suppress(Exception):
                agent._invalidate_system_prompt()
        if hasattr(agent, "_last_flushed_db_idx"):
            with contextlib.suppress(Exception):
                agent._last_flushed_db_idx = len(self.conversation_history)
        if hasattr(agent, "_session_messages"):
            agent._session_messages = self.conversation_history
        if hasattr(agent, "_db_flush_scan_prefix"):
            agent._db_flush_scan_prefix = self.conversation_history[:]

    def retry_last(self):
        """Retry the last user message: drop the last exchange and return the text to re-send
        (None when there is nothing to retry)."""
        if not self.conversation_history:
            print(t("cli.session.retry_no_messages"))
            return None

        from agent.context_compressor import (
            history_before_user_originated_turn, retryable_user_text)
        from agent.memory_manager import sanitize_context

        warm_history = list(self.conversation_history)
        user_indices = _user_turn_indices(warm_history)
        if not user_indices:
            print(t("cli.session.retry_no_user_message"))
            return None

        # Resolve a lossless live payload before touching persistence or memory. A
        # force-user-leading compaction row is one physical carrier: its handoff stays in
        # the prefix while only the embedded human ask is retried. Media cannot be replayed
        # by /retry, so fail closed before archiving anything.
        try:
            truncated, live_view = history_before_user_originated_turn(
                warm_history, user_indices[-1])
            live_content = live_view.get("content")
            if isinstance(live_content, str):
                live_content = sanitize_context(live_content).strip()
            last_message = retryable_user_text(live_content)
        except ValueError as exc:
            print(t("cli.session.retry_unsafe", error=exc))
            return None

        # Persist the rewind before publishing the shorter in-memory view: the DB owns the
        # physical carrier split so archived original + retained scaffold commit atomically.
        if self._session_db is not None and self.session_id:
            try:
                truncated = self._session_db.rewind_user_turn(
                    self.session_id, -1, warm_history=warm_history, require_retryable=True).prefix
            except Exception as exc:
                print(t("cli.session.retry_rewind_failed", error=exc))
                return None

        self._publish_truncated_history(truncated, invalidate_prompt=False)
        print(t("cli.session.retrying", preview=f"{last_message[:60]}{'...' if len(last_message) > 60 else ''}"))
        return last_message

    def undo_last(self, n: int = 1, prefill: bool = True):
        """Back up N user turns: truncate history, soft-delete on disk, prefill the composer.

        Discards everything from the Nth-from-last user message onward (clamped to the oldest
        turn). Rows are soft-deleted in SessionDB (``active=0``, kept for audit), memory
        providers get ``on_session_switch(rewound=True)``, and the agent is patched like
        /branch does. ``prefill=False`` is for programmatic callers (checkpoint rollback)
        that must not touch the input buffer. Returns the number of user turns undone (None when
        nothing changed).
        """
        from cli import logger
        if not self.conversation_history:
            print(t("cli.session.undo_no_messages"))
            return
        n = max(n, 1)

        from agent.context_compressor import history_before_user_originated_turn

        warm_history = list(self.conversation_history)
        user_indices = _user_turn_indices(warm_history)
        if not user_indices:
            print(t("cli.session.undo_no_user_message"))
            return

        turns_undone = min(n, len(user_indices))
        target_ordinal = len(user_indices) - turns_undone
        cut_idx = user_indices[target_ordinal]
        removed_count = len(warm_history) - cut_idx
        truncated, live_view = history_before_user_originated_turn(warm_history, cut_idx)
        removed_text = self._undo_content_to_text(live_view.get("content"))

        rewound_rows = 0
        if self._session_db is not None and self.session_id:
            try:
                outcome = self._session_db.rewind_user_turn(
                    self.session_id, target_ordinal, warm_history=warm_history)
                truncated = outcome.prefix
                # Canonical editable prefill: the raw carrier holds the reference-summary wrapper.
                removed_text = outcome.live_text or removed_text
                rewound_rows = outcome.rewound_count
            except Exception as e:
                logger.debug("undo: durable rewind failed: %s", e)
                print(t("cli.session.undo_failed", error=e))
                return

        # Publish only after the durable rewind succeeds (or no store exists).
        self._publish_truncated_history(truncated, invalidate_prompt=True)
        # Same hook /branch fires; rewound=True invalidates per-turn document caches.
        _mm = getattr(self.agent, "_memory_manager", None)
        # See #21910, #6672.
        if _mm is not None and self.session_id:
            with contextlib.suppress(Exception):
                _mm.on_session_switch(self.session_id, parent_session_id="", reset=False, rewound=True)

        key = "cli.session.undo_ok_one" if turns_undone == 1 else "cli.session.undo_ok_other"
        print(t(key, count=turns_undone, messages=rewound_rows or removed_count,
                backup=f"{removed_text[:60]}{'...' if len(removed_text) > 60 else ''}"))
        print(f"  {t('cli.session.undo_remaining', count=len(self.conversation_history))}")
        # Editable, not auto-sent (Claude-Code-style).
        if prefill and removed_text:
            self._prefill_input_buffer(removed_text)
        return turns_undone

    @staticmethod
    def _undo_content_to_text(content) -> str:
        """Flatten message content (str or content-part list) to plain text."""
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts = [
                p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text"
            ]
            return "\n".join(t for t in parts if t)
        return ""

    def _write_terminal_breadcrumb(self) -> None:
        """Record this terminal's live session for bare ``hermes -c``. Called whenever
        ``self.session_id`` is (re)assigned so a later bare ``-c`` in THIS terminal resumes
        this conversation's live tip. Best-effort; no-op without a terminal identity."""
        with contextlib.suppress(Exception):
            from hermes_cli.terminal_breadcrumbs import write_breadcrumb

            write_breadcrumb(self.session_id)

    def _transfer_session_yolo(self, old_session_id: str, new_session_id: str) -> None:
        """Move YOLO bypass state to a new session key when ``self.session_id`` is reassigned
        mid-run (/branch, auto-compression rotation) — ``_session_yolo`` is keyed by id, so
        without this the toggle silently reverts. Mirrors tui_gateway's rename path."""
        if not old_session_id or not new_session_id or old_session_id == new_session_id:
            return
        try:
            from tools.approval import (
                disable_session_yolo, enable_session_yolo, is_session_yolo_enabled)
        except Exception:
            return
        if is_session_yolo_enabled(old_session_id):
            enable_session_yolo(new_session_id)
            disable_session_yolo(old_session_id)
            # Carry the persisted flag onto the continuation row so a later --resume restores
            # it too. getattr: tests call this unbound against a minimal stand-in.
            _persist = getattr(self, "_persist_session_yolo", None)
            if _persist:
                _persist(new_session_id, True)

    def _is_session_yolo_active(self) -> bool:
        """Whether YOLO bypass is on for this session: reads ``tools.approval._session_yolo``
        (not a stale env var) and honors the frozen process-start ``--yolo`` flag."""
        try:
            from tools.approval import _YOLO_MODE_FROZEN, is_session_yolo_enabled
        except Exception:
            return False
        if _YOLO_MODE_FROZEN:
            return True
        # getattr: __new__-built test fixtures skip __init__; the status-bar builders
        # swallow exceptions but would lose every field after the failure.
        return is_session_yolo_enabled(getattr(self, "session_id", None) or "default")

    def _toggle_yolo(self):
        """Toggle per-session YOLO mode (skip dangerous-command approvals).

        Mirrors the gateway/TUI ``/yolo`` handlers. Deliberately does NOT touch
        ``HERMES_YOLO_MODE``: that env var is frozen into ``tools.approval._YOLO_MODE_FROZEN``
        at import (so prompt-injected skills can't flip the bypass), making a later set a
        silent no-op. ``run_conversation`` binds ``self.session_id`` as the active approval
        key, so the bypass applies to the very next dangerous command.
        """
        from cli import _cprint
        from hermes_cli.colors import Colors as _Colors
        from tools.approval import (
            _YOLO_MODE_FROZEN, disable_session_yolo, enable_session_yolo, is_session_yolo_enabled)

        # A frozen process-level bypass short-circuits the approval gate ahead of the session
        # check — toggling "OFF" would be a false safety claim. Say so instead.
        if _YOLO_MODE_FROZEN:
            state = f"{_Colors.BOLD}{_Colors.RED}{t('cli.session.yolo_state_locked_on')}{_Colors.RESET}"
            _cprint(f"  {t('cli.session.yolo_locked', state=state)}")
            return

        session_key = self.session_id or "default"
        # getattr: tests call this unbound against a minimal stand-in; persistence is best-effort.
        _persist = getattr(self, "_persist_session_yolo", None)
        if is_session_yolo_enabled(session_key):
            disable_session_yolo(session_key)
            if _persist:
                _persist(session_key, False)
            state = f"{_Colors.BOLD}{_Colors.RED}{t('cli.shared.label_off_upper')}{_Colors.RESET}"
            _cprint(f"  {t('cli.session.yolo_off', state=state)}")
        else:
            enable_session_yolo(session_key)
            if _persist:
                _persist(session_key, True)
            state = f"{_Colors.BOLD}{_Colors.GREEN}{t('cli.shared.label_on_upper')}{_Colors.RESET}"
            _cprint(f"  {t('cli.session.yolo_on', state=state)}")

    def _persist_session_yolo(self, session_key: str, enabled: bool) -> None:
        """Persist the YOLO flag to the session row so --resume restores it. Best-effort; the
        in-memory toggle is authoritative. Skipped without a store or before the row exists
        (rows are created lazily on the first turn)."""
        db = getattr(self, "_session_db", None)
        if db is None or not session_key or session_key == "default":
            return
        with contextlib.suppress(Exception):
            db.set_session_yolo(session_key, enabled)

    def _manual_compress(self, cmd_original: str = ""):
        """Manually trigger context compression.

        * ``/compress [<focus>]`` — compress the whole history; an optional focus topic tells
          the summariser what to preserve while discarding the rest more aggressively.
        * ``/compress here [N]`` — boundary-aware: summarize everything except the most recent
          ``N`` exchanges (default 2), kept verbatim.
        * ``--preview`` reports what would happen and changes nothing.
        No ``compression_enabled`` gate: that flag disables *automatic* compaction only, and
        the context-overflow error path directs users here when it is off.
        """
        from agent.conversation_compression import finalize_context_engine_compression_notification
        from agent.conversation_compression_manual import (
            AGGRESSIVE_UNSUPPORTED, MIN_MESSAGES, compress_now, parse_compress_args, render_compress_result)

        if len(self.conversation_history or ()) < MIN_MESSAGES:
            print(t("cli.session.compress_too_short", min=MIN_MESSAGES))
            return
        if not self.agent:
            print(t("cli.session.compress_no_agent"))
            return
        _parts = (cmd_original or "").strip().split(None, 1)
        request = parse_compress_args(_parts[1] if len(_parts) > 1 else "")
        if request.aggressive:
            print(f"(._.) {AGGRESSIVE_UNSUPPORTED}")
            if not request.preview:
                return
        if request.preview:
            for line in render_compress_result(compress_now(self.agent, self.conversation_history, request)):
                print(f"🗜️  {line}")
            return

        original_count = len(self.conversation_history)
        with self._busy_command(t("cli.session.compress_busy"), blocks_input=False):
            try:
                if request.partial:
                    print(t("cli.session.compress_summarizing", count=original_count, keep_last=request.keep_last))
                elif request.focus_topic:
                    print(t("cli.session.compress_focus", count=original_count, focus=request.focus_topic))
                else:
                    print(t("cli.session.compress_plain", count=original_count))
                result = compress_now(self.agent, self.conversation_history, request,
                                      task_id=self.session_id or "default")
                if result.status != "compressed":
                    for line in render_compress_result(result):
                        print(f"  {line}")
                    return
                self.conversation_history = result.after_messages
                # _compress_context ends the old session and creates a child session on the
                # agent. Sync the CLI's session_id so /status, /resume, exit summary and title
                # generation point at the live continuation, not the ended parent.
                agent_sid = getattr(self.agent, "session_id", None)
                if agent_sid and agent_sid != self.session_id:
                    self.session_id = self.agent.session_id
                    self._write_terminal_breadcrumb()
                    self._pending_title = None
                    # Persist the new handoff from offset 0 so resume can recover it after exit.
                    self.agent._flush_messages_to_session_db(self.conversation_history, None)
                finalize_context_engine_compression_notification(self.agent, committed=True)
                summary = result.summary
                if summary.get("aborted") or summary.get("fallback_used") or summary.get("refused_would_grow"):
                    icon = "⚠️"
                else:
                    icon = "🗜️" if summary["noop"] else "✅"
                print(f"  {icon} {summary['headline']}")
                print(f"     {summary['token_line']}")
                if summary["note"]:
                    print(f"     {summary['note']}")
            except Exception as e:
                finalize_context_engine_compression_notification(self.agent, committed=False)
                print(f"  {t('cli.session.compress_failed', error=e)}")

    def _persist_prompt_summary(self, icon: str, label: str, detail: str, outcome: str) -> None:
        """Print a one-line scrollback summary of a resolved modal prompt (approval/clarify
        panels vanish on repaint); gated by ``display.persist_prompts``."""
        from cli import CLI_CONFIG, _DIM, _RST, _cprint
        if not CLI_CONFIG.get("display", {}).get("persist_prompts", True):
            return
        detail, outcome = (_squash(s) for s in (detail, outcome))
        _cprint(f"\n{_DIM}{t('cli.session.persist_prompt_summary', icon=icon, label=label, detail=detail, outcome=outcome)}{_RST}")

    def _clear_terminal_on_exit(self):
        """Clear screen + scrollback (``ESC[3J ESC[2J ESC[H``) so nothing is stranded above
        the exit summary. Only safe after ``app.run()`` returned and prompt_toolkit restored
        terminal modes. Skips when stdout isn't a console; falls back to ``clear``/``cls``."""
        try:
            stream = sys.stdout
            if stream is None or not stream.isatty():
                return
        except Exception:
            return
        try:
            stream.write("\033[3J\033[2J\033[H")
            stream.flush()
        except Exception:
            # Fallback for terminals that reject the escape sequence. Never os.system():
            # it spawns a cmd.exe/shell window that flashes on Windows (#116904) and a
            # minimal container without `clear` on PATH just no-ops through the shell.
            try:
                import subprocess

                from hermes_cli._subprocess_compat import windows_hide_flags

                if os.name == "nt":
                    argv = ["cmd", "/c", "cls"]  # `cls` is a cmd builtin, not an exe
                else:
                    clear_bin = shutil.which("clear")
                    argv = [clear_bin] if clear_bin else []
                if argv:
                    subprocess.run(
                        argv,
                        stdin=subprocess.DEVNULL,
                        creationflags=windows_hide_flags(),
                        check=False,
                    )
            except Exception:
                pass

    def _persist_active_session_before_close(self):
        """Best-effort flush of the agent's live ``_session_messages`` before ``end_session()``
        — a terminal close/SIGHUP can unwind the app while the agent thread still holds the
        current turn only in memory."""
        from cli import logger
        agent = getattr(self, "agent", None)
        if not agent or not hasattr(agent, "_persist_session"):
            return

        persist_lock = getattr(agent, "_session_persist_lock", None)

        def _snapshot_and_persist() -> None:
            # Must share the staging lock with ``chat()``: otherwise close can retain a
            # history baseline just before chat appends its pending dict, and the later flush
            # stamps that dict durable without writing a row.
            messages = getattr(agent, "_session_messages", None)
            pending_cli_message = getattr(agent, "_pending_cli_user_message", None)
            if not isinstance(messages, list):
                messages = getattr(self, "conversation_history", None)
            if not isinstance(messages, list):
                return
            if isinstance(pending_cli_message, dict) and not any(
                m is pending_cli_message for m in messages):
                # The UI accepted a new input but the worker still exposes its prior snapshot.
                messages = [*messages, pending_cli_message]
            if not messages:
                return

            # Baseline: the CLI history a normal turn built its new list from, so a signal
            # between assigning ``_session_messages`` and the DB flush cannot append the
            # durable prefix twice. When both names alias the same live list, marker-only
            # persistence would mark an unflushed tail durable — pass None instead.
            conversation_history = getattr(self, "conversation_history", None)
            if (
                isinstance(conversation_history, list)
                and conversation_history
                and conversation_history[-1] is pending_cli_message):
                # Accepted but not yet durable: exclude it from the resumed-history baseline.
                conversation_history = conversation_history[:-1]
            elif not isinstance(conversation_history, list) or conversation_history is messages:
                conversation_history = None

            # A first-turn close can precede the cached prompt; build it so the durable
            # transcript never gets a NULL system_prompt cache entry.
            if getattr(agent, "_cached_system_prompt", None) is None:
                try:
                    from agent.conversation_loop import _restore_or_build_system_prompt

                    _restore_or_build_system_prompt(agent, None, conversation_history)
                except Exception:
                    logger.debug("Could not build system prompt during CLI close", exc_info=True)
                    return
            if getattr(agent, "_cached_system_prompt", None) is None:
                return

            agent._ensure_db_session()
            agent._persist_session(messages, conversation_history)
            if getattr(agent, "session_id", None):
                self.session_id = agent.session_id
                self._write_terminal_breadcrumb()

        try:
            # Create the DB session row now that _cached_system_prompt is populated, so the persisted
            # snapshot is written non-NULL on the first turn (Issue #45499). Idempotent:
            # _ensure_db_session() no-ops once the row exists. Must run BEFORE preflight compression:
            # in-place compaction inserts message rows referencing this session (archive_and_compact), and
            # rotation creates a child with parent_session_id pointing at it — with PRAGMA foreign_keys=ON,
            # a missing parent row fails both INSERTs on a fresh oversized first turn. The user-turn crash
            # persist itself runs LATER (after memory prefetch / pre_llm_call), so the row is written once
            # with its final api_content — both steps take the same per-agent persist lock as CLI close
            # persistence.
            if persist_lock is None:
                _snapshot_and_persist()
            else:
                with persist_lock:
                    _snapshot_and_persist()
        except (Exception, KeyboardInterrupt) as e:
            logger.debug("Could not persist active CLI session before close: %s", e)

    def _print_exit_summary(self, clear_screen: bool = True):
        """Print session resume info on exit. ``clear_screen`` (interactive TUI teardown)
        wipes screen + scrollback first; single-query mode passes False to keep the answer.

        Args: clear_screen: When True (default), clear the terminal screen and scrollback before printing
        the summary. See #38252, #53009.
        """
        from cli import datetime
        if clear_screen:
            # Clear the screen + scrollback before printing the summary so the live bottom chrome (status
            # bar, input box, separator rules) and the rest of the session transcript don't get stranded
            # above the exit summary (#38252). By this point app.run() has returned and prompt_toolkit has
            # restored terminal modes, so writing raw escapes to stdout is safe. ESC[3J clears scrollback,
            # ESC[2J clears the visible screen, ESC[H homes the cursor — so the summary prints at a clean
            # top-left. Falls back to the platform clear command if stdout isn't a TTY-capable stream.
            # Honors NO_COLOR/dumb terminals by skipping silently when there's no real console.
            self._clear_terminal_on_exit()
        print()
        msg_count = len(self.conversation_history)
        if not msg_count:
            try:
                from hermes_cli.skin_engine import get_active_goodbye
                goodbye = get_active_goodbye(t("cli.session.goodbye"))
            except Exception:
                goodbye = t("cli.session.goodbye")
            print(goodbye)
            return

        user_msgs = len([m for m in self.conversation_history if m.get("role") == "user"])
        tool_calls = len([
            m for m in self.conversation_history if m.get("role") == "tool" or m.get("tool_calls")])
        elapsed = datetime.now() - self.session_start
        hours, remainder = divmod(int(elapsed.total_seconds()), 3600)
        minutes, seconds = divmod(remainder, 60)
        duration_str = t("cli.shared.duration_s", seconds=seconds)
        if hours > 0:
            duration_str = t("cli.shared.duration_hm_rest", hours=hours, minutes=minutes, rest=duration_str)
        elif minutes > 0:
            duration_str = t("cli.shared.duration_m_rest", minutes=minutes, rest=duration_str)

        session_title = None
        if self._session_db:
            with contextlib.suppress(Exception):
                session_title = self._session_db.get_session_title(self.session_id)

        print(t("cli.session.exit_resume_hint"))
        # Session IDs are profile-constrained: non-default profiles need `-p <profile>` in
        # the hint ("default"/"custom" use the standard HERMES_HOME).
        try:
            from hermes_cli.profiles import get_active_profile_name
            _active_profile = get_active_profile_name()
        except Exception:
            _active_profile = "default"
        profile_flag = "" if _active_profile in ("default", "custom") else f" -p {_active_profile}"
        print(f"  hermes --resume {self.session_id}{profile_flag}")
        if session_title:
            print(f"  hermes -c \"{session_title}\"{profile_flag}")
        print()
        print(t("cli.session.exit_label_session", session_id=self.session_id))
        if session_title:
            print(t("cli.session.exit_label_title", title=session_title))
        print(t("cli.session.exit_label_duration", duration=duration_str))
        print(t("cli.session.exit_label_messages", count=msg_count, user=user_msgs, tool_calls=tool_calls))