nousresearch-hermes-agent-32018-33241

drwxr-xr-x 1 root root   4096 Nov  5 07:30 .plans
-rw-r--r-- 1 root root  53040 Nov  5 07:30 AGENTS.md
-rw-r--r-- 1 root root  44350 Nov  5 07:30 CONTRIBUTING.md
-rw-r--r-- 1 root root  12197 Nov  5 07:30 Dockerfile
-rw-r--r-- 1 root root   1070 Nov  5 07:27 LICENSE
-rw-r--r-- 1 root root     87 Nov  5 07:30 M
…[6205 chars omitted]…
./agent/retry_utils.py
./agent/shell_hooks.py
./agent/skill_bundles.py
./agent/skill_commands.py
./agent/skill_preprocessing.py
./agent/skill_utils.py
./agent/stream_diag.py
./agent/subdirectory_hints.py
./agent/system_prompt.py
./agent/think_scrubber.py
./agent/title_generator.py
./agent/tool_dispatch_helpers.py
./agent/tool_executor.py
./agent/tool_guardrails.py
./agent/tool_result_classification.py

def _bound(fn, instance):
    """Bind an unbound method to a stand-in instance."""
    return fn.__get__(instance, type(instance))


def _make_self(prompt_response):
    """Build a minimal stand-in 'self' for _confirm_destructive_slash."""
    from cli import HermesCLI

    self_ = SimpleNamespace(
        _app=None,
        _prompt_text_input=lambda _prompt: prompt_response,
        _prompt_text_input_modal=lambda **_kw: prompt_response,
    )
    self_._normalize_slash_confirm_choice = _bound(
        HermesCLI._normalize_slash_confirm_choice, self_,
    )
    
…[8628 chars omitted]…
it_destructive_skip = HermesCLI._split_destructive_skip

    with patch(
        "cli.load_cli_config",
        return_value={"approvals": {"destructive_slash_confirm": True}},
    ):
        result = _bound(HermesCLI._confirm_destructive_slash, self_)(
            "new", "detail", cmd_original="/new My Session",
        )

    # Prompt was reached and returned cancel → None.
    assert result is None

        prompt appeared above the TUI, choices were redrawn later, and Enter
        could be interpreted as EOF/exit.  A first-class modal state keeps the
        choices visible and lets the normal Enter key binding submit the typed
        or highlighted choice.

        **Platform note (Windows dead-lock — issue #30768):**
        The queue-based modal relies on prompt_toolkit key bindings receiving
        keyboard events and calling ``_submit_slash_confir
…[8629 chars omitted]…
     }
        self._invalidate(min_interval=0.0)

    def _close_model_picker(self) -> None:
        self._model_picker_state = None
        self._restore_modal_input_snapshot()
        self._invalidate(min_interval=0.0)

    @staticmethod
    def _compute_model_picker_viewport(
        selected: int,
        scroll_offset: int,
        n: int,
        term_rows: int,
        reserved_below: int = 6,

        assert _strip_leaked_bracketed_paste_wrappers(text) == "hello"

    def test_strips_visible_caret_escape_wrappers(self):
        text = "^[[200~hello^[[201~"
        assert _strip_leaked_bracketed_paste_wrappers(text) == "hello"

    def test_strips_degraded_bracket_only_wrappers(self):
        text = "[200~hello[201~"
        assert _strip_leaked_bracketed_paste_wrappers(text) == "hello"

    def test_strips_degraded_bracket_only_wrappers_after_whitespace(self):
        text 
…[930 chars omitted]…
le_stripping_wrappers(self):
        text = "^[[200~line 1\nline 2\nline 3^[[201~"
        assert _strip_leaked_bracketed_paste_wrappers(text) == "line 1\nline 2\nline 3"

    def test_preserves_multiline_content_while_stripping_degraded_bracket_only_wrappers(self):
        text = "[200~line 1\nline 2\nline 3[201~"
        assert _strip_leaked_bracketed_paste_wrappers(text) == "line 1\nline 2\nline 3"

4. No-app path still uses the stdin fallback (existing behavior)
5. Empty choices returns None (existing behavior)
"""

import queue
import sys
import threading
import time
from unittest.mock import MagicMock, patch

import pytest


def _make_cli():
    """Minimal H
…[8628 chars omitted]…
"3"), \
             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
            result = cli._confirm_destructive_slash(
                "reset",
                "This starts a fresh session.\nThe current conversation history will be discarded.",
            )

        # Choice "3" normalizes to "cancel", which returns None.
        assert result is None

            style = 'class:approval-selected' if i == selected else 'class:approval-choice'
            _append_panel_line(lines, 'class:approval-border', style, wrapped, box_width)

        if desc_wrapped:
            if not use_compact_chrome:
                _append_blank_pane
…[7167 chars omitted]…
S output is enabled
        self._voice_recorder = None     # AudioRecorder instance (lazy init)
        self._voice_recording = False   # Whether currently recording
        self._voice_processing = False  # Whether STT is in progress
        self._voice_continuous = False  # Whether to auto-restart after agent responds
        self._voice_tts_done = threading.Event()  # Signals TTS playback finished

"""

# IMPORTANT: hermes_bootstrap must be the very first import — UTF-8 stdio
# on Windows.  No-op on POSIX.  See hermes_bootstrap.py for full rationale.
try:
    import hermes_bootstrap  # noqa: F401
except ModuleNotFoundError:
    # Graceful fallback when hermes_bootstrap isn't registered in the venv
    # yet — happens during partial ``hermes updat
…[4890 chars omitted]…
ser_debug_ready,
    manual_chrome_debug_command,
    try_launch_chrome_debug,
)
from hermes_cli.env_loader import load_hermes_dotenv
from utils import base_url_host_matches, is_truthy_value

_hermes_home = get_hermes_home()
_project_env = Path(__file__).parent / '.env'
load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)


_REASONING_TAGS = (
    "REASONING_SCRATCHPAD",
    "think",

2531:    """Rich Console adapter for prompt_toolkit's patch_stdout context.
2720:    """Return plugin command names (without slash prefix) for dispatch matchi
…[2463 chars omitted]…
   _pt_renderer._output_screen_diff = _patched_output_screen_diff
14150:                _pt_renderer._hermes_osd_patched = True
14293:                        # Dispatch to a daemon thread so play_beep (sd.wait) and
14493:        # Run the application with patch_stdout for proper output handling
14495:            with patch_stdout():
14504:                    pass  # No running loop -- nothing to patch

            loop.default_exception_handler(context)

        # Vali
…[3828 chars omitted]…
making
            # API calls and exits promptly (agent_thread is daemon, so the
            # process will exit once the main thread finishes, but interrupting
            # avoids wasted API calls and lets run_conversation clean up).
            if self.agent and getattr(self, '_agent_running', False):
                try:
                    self.agent.interrupt()
                except Exception:

        i = Vt100Parser(callback)
        i.feed('data\x01...')

    :attr feed_key_callback: Function that will be called when a key is parsed.
    """

    # Lookup table of ANSI escape sequences for a VT100 terminal
    # Hint: in order to know what sequences your terminal writes to stdin, run
    #       "od -c" and start typing.
    def __init__(self, feed_key_callback: Callable[[KeyPress], None]) -> None:
        self.feed_key_callback = feed_key_callback
        self.reset()

    def reset(self, request: bool = False) -> None:
        self._in_bracketed_paste = False
      
…[5263 chars omitted]…
 else. This flush function should be called after a
        timeout, and processes everything that's still in the buffer as-is, so
        without assuming any characters will follow.
        """
        self._input_parser.send(_Flush())

    def feed_and_flush(self, data: str) -> None:
        """
        Wrapper around ``feed`` and ``flush``.
        """
        self.feed(data)
        self.flush()

from ..keys import Keys
from .ansi_escape_sequences import ANSI_SEQUENCES

__all__ = [
    "Vt100Parser",
]


# Regex matching any CPR response
# (Note that we use '\Z' instead of '$', because '$' could include a trailing
# newline.)
_cpr_response_re = re.compile("^" + re.escape("\x1b[") + r"\d+;\d+R\Z")

# Mouse events:
# Typical: "Esc[MaB*"  Urxvt: "Esc[96;14;13M" and for Xterm SGR: "Esc[<64;85;12M"
_mouse_event_re = re.compile("^" + re.escape("\x1b[") + r"(<?[\d;]+[mM]|M...)\Z")

# Regex matching any valid prefix of a CPR response.
# (Note that it doesn't contain the last character, the 'R'. The prefix has to
# be shorter.)
_cpr_response_prefix_re = re.compile("^" + re.escape("\x1b[") + r"[\d;]*\Z")

_mouse_event_prefix_re = re.compile("^" + re.escape("\x1b[") + r"(<?[\d;]*|M.{0,2})\Z")


class _Flush:
    """Helper object to indicate flush operation to the parser."""

    pass


class _IsPrefixOfLongerMatchCache(Dict[str, bool]):
    """
    Dictionary that maps input sequences to a boolean indicatin

    def test_main_thread_non_windows_uses_modal(self):
        """On macOS/Linux main thread, the queue-based modal is still used."""
        cli = _make_cli()

        # We need to simulate the modal receiving a response. We'll patch
        # the response_queue to immediately return a value.
        with patch.object(sys, "platform", "darwin"), \
             patch.object(cli, "_capture_modal_input_snapshot"), \
       
…[6263 chars omitted]…
from cli import HermesCLI

    self_ = _make_self(prompt_response="maybe")

    with patch(
        "cli.load_cli_config",
        return_value={"approvals": {"destructive_slash_confirm": True}},
    ):
        result = _bound(HermesCLI._confirm_destructive_slash, self_)(
            "clear", "detail",
        )

    assert result is None


def test_gate_on_choice_always_persists_and_returns_always():

            if not use_compact_chrome:
                _append_blank_panel_line(lines, 'class:approval-border', box_width)
            for wrapped in desc_wrapped:
               
…[3965 chars omitted]…
 picker modal ---
            if self._model_picker_state:
                try:
                    self._handle_model_picker_selection()
                except Exception as _exc:
                    _cprint(f"  ✗ Model selection failed: {_exc}")
                    self._close_model_picker()
                event.app.current_buffer.reset()
                event.app.invalidate()
                return

        self._status_bar_suppressed_after_resize = False
        self._resize_recovery_lock = threading.Lock()
        self._resize_recovery_timer = None
        self._resize_recovery_pending = False

        # Background task tracking: {task_id: threading.
…[5173 chars omitted]…
p.current_buffer.reset()
                event.app.invalidate()
                return

            # Cancel clarify prompt
            if self._clarify_state:
                self._clarify_state["response_queue"].put(
                    "The user cancelled. Use your best judgement to proceed."
                )
                self._clarify_state = None
                self._clarify_freetext = False

900         self, loop: AbstractEventLoop, context: dict[str, Any]
917                 print("\nUnhandled exception in event loop:")
1046         loop = self.loop or get_running_loop()

    return text


# Cursor Position Report (CPR / DSR) response, format ``ESC[<row>;<col>R``.
# prompt_toolkit's _on_resize() + renderer send ``ESC[6n`` queries to the
# terminal; under resize storms or tab switches the terminal's reply can
# race past the input parser and end up in the input buffer as literal
# text (see issue #14692). Also matches the visible-form ``^[[<row>;<col>R``
#
…[7393 chars omitted]…
e whenever emitted text ends with '\n'.
        if not hasattr(self, "_stream_last_was_newline"):
            self._stream_last_was_newline = True  # start of stream = boundary

        if not getattr(self, "_in_reasoning_block", False):
            for tag in _OPEN_TAGS:
                search_start = 0
                while True:
                    idx = self._stream_prefilt.find(tag, search_start)

    except Exception:
        pass


def _strip_leaked_terminal_responses_with_meta(text: str) -> tuple[str, bool]:
    """Strip leaked terminal control-response sequences from user input.

    Covers Cursor Position Report (CPR / DSR) responses — ``ESC[<row>;<col>R``
    and the visible ``^[[<row>;<col>R`` form. These are replies the terminal
    sends back to queries prompt_toolkit makes during ``_on_resize`` /
    ``_re
…[347 chars omitted]…
had_mouse_reports)``
    so callers can trigger an in-place terminal mode recovery when needed.
    """
    if not text:
        return text, False

    has_esc = "\x1b[" in text
    has_visible = "^[" in text
    has_bare_mouse = "<" in text and ";" in text and ("M" in text or "m" in text)
    if not (has_esc or has_visible or has_bare_mouse):
        return text, False

    had_mouse_reports = False

    // continuation of the buffered sequence (e.g. `[<64;74;16M` after a
    // lone ESC). Node's event loop runs the timers phase before the poll
    // phase, so when a heavy render blocks the loop past 50ms, this timer
    // fires before the queued readable event even though the bytes are
    // already buffered. Re-arm instead of flushing: handleReadable will
    // drain stdin next and clear this timer
…[1978 chars omitted]…
set DEC private modes; re-assert mouse
    // tracking. Checked before the read loop so one Date.now() covers
    // all chunks in this readable event.
    const now = Date.now()

    if (now - this.lastStdinTime > STDIN_RESUME_GAP_MS) {
      this.props.onStdinResume?.()
    }

    this.lastStdinTime = now

tests/cli/test_cli_bracketed_paste_sanitizer.py:6:class TestStripLeakedBracketedPasteWrappers:

        if self.loop is None or self.loop.is_closed():
            return

        # Never schedule a second redraw, when a previous one has not yet been
        # executed. (This should protect against other threads calling
        # 'invalidate' many times, resulting in 100% CPU.)
        if self._invalidated:
            return
        else:
            self._invalidated = True
…[628 chars omitted]…
           async def redraw_in_future() -> None:
                    await sleep(cast(float, self.min_redraw_interval) - diff)
                    schedule_redraw()

                self.loop.call_soon_threadsafe(
                    lambda: self.create_background_task(redraw_in_future())
                )
            else:
                schedule_redraw()
        else:
            schedule_redraw()

        from hermes_cli.debug import run_debug_share
        from types import SimpleNamespace

        args = SimpleNamespace(lines=200, expire=7, local=False)
        run_debug_share(args)

    def _handle_update_command(self) -> bool:
        """Handle /update — update Hermes Agent to the latest version.

        In the classic CLI this exits the session and relaunches as
        ``her
…[2276 chars omitted]…
gent = self.agent
        calls = agent.session_api_calls

        if calls == 0:
            print("(._.) No API calls made yet in this session.")
            return

        # ── Rate limits (shown first when available) ────────────────
        rl_state = agent.get_rate_limit_state()
        if rl_state and rl_state.has_data:
            from agent.rate_limit_tracker import format_rate_limit_display

                continue
            kept.append(tok)
        return " ".join(kept), skip

    def _confirm_destructive_slash(
        self,
        command: str,
        detail: str,
        cmd_original: Optional[str] = None,
    ) -> Optional[str]:
        """Prompt the user to confirm a destructive session slash command.

        Used by ``/clear``, ``/new``/``/reset``, and ``/undo`` before they
        discard conversation state.  Three-option prompt:

    
…[2931 chars omitted]…
= "always":
            if save_config_value("approvals.destructive_slash_confirm", False):
                print("🔒 Future /clear, /new, /reset, and /undo will run without confirmation.")
                print("   Re-enable via `approvals.destructive_slash_confirm: true` in config.yaml.")
            else:
                print("⚠️  Couldn't persist opt-out — proceeding once.")

        return choice

    arrived input is parsed through the ordinary path.

    The helper is deliberately self-contained.  The regression tests extract
    this function's source and exec
…[2480 chars omitted]…
t extend the deadline.
            if not was_in_paste or getattr(self, "_hermes_bp_start", None) is None:
                self._hermes_bp_start = time.monotonic()
        else:
            self._hermes_bp_start = None

    _feed_with_bracketed_paste_timeout._hermes_bp_timeout_wrapper = True
    _Vt100Parser.feed = _feed_with_bracketed_paste_timeout
    _vt100_parser._hermes_bp_timeout_patched = True

    def test_show_banner_does_not_print_skills():
        """show_banner() no longer prints the activated skills line — it moved to run()."""
…[8679 chars omitted]…
:TestGatewayPersonalityNone::test_default_clears_ephemeral_prompt
FAILED tests/cli/test_personality_none.py::TestGatewayPersonalityNone::test_list_includes_none
FAILED tests/cli/test_personality_none.py::TestGatewayPersonalityNone::test_unknown_shows_none_in_available
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 5 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
5 failed, 528 passed, 11 warnings in 228.76s (0:03:48)

+
+    prompt_toolkit's ``Vt100Parser`` buffers everything after ``ESC[200~``
+    until it sees ``ESC[201~``.  If that end marker is lost (SSH interruption,
+    terminal race, sleep/wake, or a torn write), every later keystroke is
+    treated as paste content and the prompt appears permanently frozen.
+
+    This installs a small wrapper around ``Vt100Parser.feed``.  When a feed
+    arrives more than two seconds after paste mode was entered and no end
+
…[8629 chars omitted]…
slash_confirm_deadline = 0
-                self._restore_modal_input_snapshot()
-                self._invalidate()
-        return None
+            # Clear state from the requesting thread too, so callers see a
+            # clean slate immediately even if the event-loop callback is
+            # briefly delayed.  Restore the draft on the loop thread.
+            self._slash_confirm_state = None

+    until it sees ``ESC[201~``.  If that end marker is lost (SSH interruption,
+    terminal race, sleep/wake, or a torn write), every later keystroke is
+    treated as paste content and the prompt appears permanently frozen.
+
+    This installs a small wrapper around ``Vt100Parser.feed``.  When a feed
+    arrives more than two seconds after paste mode was entered and no end
+    marker has completed the paste, the text already buffered is delivered as
+    a normal ``BracketedPaste`` key press, paste mode
…[8629 chars omitted]…
briefly delayed.  Restore the draft on the loop thread.
+            self._slash_confirm_state = None
+            self._slash_confirm_deadline = 0
+            try:
+                loop.call_soon_threadsafe(_close_modal)
+            except Exception:
+                _close_modal()
 
     def _submit_slash_confirm_response(self, value: str | None) -> None:
         state = self._slash_confirm_state

                    pass
            finally:
                self._status_bar_visible = was_visible
                self._app.invalidate()
        else:
            _ask()
        return result[0]

    def _prompt_text_input_modal(
        self,
        *,
        title: str,
        detail: str,
        choices: list[tuple[str, str, str]],
        timeout: float = 120,
  
…[7284 chars omitted]…
],
    ) -> str | None:
        if raw is None:
            return None
        choice_raw = raw.strip().lower()
        if not choice_raw:
            return None
        aliases = {
            "1": "once",
            "once": "once",
            "approve": "once",
            "yes": "once",
            "y": "once",
            "ok": "once",
            "2": "always",
            "always": "always",

-rw-r--r-- 1 root root   5743 Nov  2 11:22 .mailmap
drwxr-xr-x 1 root root   4096 Nov  2 11:22 .plans
-rw-r--r-- 1 root root  53040 Nov  2 11:22 AGENTS.md
-rw-r--r-- 1 root root  44350 Nov  2 11:22 CONTRIBUTING.md
-rw-r--r-- 1 root root  12197 Nov  2 11:22 Dockerfile
-rw-r--r-- 1 root root   1070 Nov  2 11:20 LICENSE
-rw-r--r-- 1 root root     87 Nov  2 1
…[8335 chars omitted]…
s_compat.py
hermes_cli/auth.py
hermes_cli/auth_commands.py
hermes_cli/azure_detect.py
hermes_cli/backup.py
hermes_cli/banner.py
hermes_cli/browser_connect.py
hermes_cli/bundles.py
hermes_cli/callbacks.py
hermes_cli/checkpoints.py
hermes_cli/claw.py
hermes_cli/cli_output.py
hermes_cli/clipboard.py
hermes_cli/codex_models.py
hermes_cli/codex_runtime_plugin_migration.py
hermes_cli/codex_runtime_switch.py

7075:        
…[8628 chars omitted]…
ts/gateway/test_telegram_thread_fallback.py:1004:    adapter._slash_confirm_state = {"confirm-1": "session-1"}
tests/hermes_cli/test_setup_prompt_menus.py:4:def test_prompt_strips_bracketed_paste_markers(monkeypatch):
tests/hermes_cli/test_setup_prompt_menus.py:15:def test_password_prompt_strips_bracketed_paste_markers(monkeypatch):
tests/tools/test_ansi_strip.py:40:    def test_bracketed_paste(self):

"""

# IMPORTANT: hermes_bootstrap must be the very first import — UTF-8 stdio
# on Windows.  No-op on POSIX.  See hermes_bootstrap.py for full rationale.
try:
    import hermes_bootstrap  # noqa: F401
except ModuleNotFoundError:
    # Graceful fallback when hermes_bootstrap isn't registered in the venv
    # yet — happens during partial ``hermes updat
…[8630 chars omitted]…
confirm the highlighted selection ---
            if self._clarify_state and not self._clarify_freetext:
                state = self._clarify_state
                selected = state["selected"]
                choices = state.get("choices") or []
                if selected < len(choices):
                    state["response_queue"].put(choices[selected])
                    self._clarify_state = None

        else:
            _pick()

        return result[0]

    def _prompt_text_input(self, prompt_text: str) -> str | None:
        """Prompt for free-text input safely inside or outside prompt_toolkit.

        Mirrors the thread-aware guard in ``_run_curses_picker``: ``run_in_terminal``
        returns a coroutine that must be awaited by the prompt_toolkit event loop,
        which only exists on th
…[8630 chars omitted]…
 else. This flush function should be called after a
        timeout, and processes everything that's still in the buffer as-is, so
        without assuming any characters will follow.
        """
        self._input_parser.send(_Flush())

    def feed_and_flush(self, data: str) -> None:
        """
        Wrapper around ``feed`` and ``flush``.
        """
        self.feed(data)
        self.flush()

        prompt appeared above the TUI, choices were redrawn later, and Enter
        could be interpreted as EOF/exit.  A first-class modal state keeps the
        choices visible and lets the normal Enter key binding submit the typed
        or highlighted choice.

        **Platform note (Windows dead-lock — issue #30768):**
        The queue-based modal relies on prompt_toolkit key bindings receiving
        keyboard events and calling ``_submit_slash_confir
…[3984 chars omitted]…
ce(
        self,
        raw: str | None,
        choices: list[tuple[str, str, str]],
    ) -> str | None:
        if raw is None:
            return None
        choice_raw = raw.strip().lower()
        if not choice_raw:
            return None
        aliases = {
            "1": "once",
            "once": "once",
            "approve": "once",
            "yes": "once",
            "y": "once",

            # Received ANSI sequence that corresponds with multiple keys
            # (probably alt+something). Handle keys individually, but only pass
            # data payload to first KeyPress (so that we won't insert it
            # multiple times).
            for i, k in enumerate(key):
                self._call_handler(k, insert_text if i == 0 else "")
        else:
            if key == Keys.BracketedPaste:
                self._in_bracketed_paste = True
                self._paste_buffer = "
…[8629 chars omitted]…
):
    """No input (None / EOF / Ctrl-C) treated as cancel."""
    from cli import HermesCLI

    self_ = _make_self(prompt_response=None)

    with patch(
        "cli.load_cli_config",
        return_value={"approvals": {"destructive_slash_confirm": True}},
    ):
        result = _bound(HermesCLI._confirm_destructive_slash, self_)(
            "clear", "detail",
        )

    assert result is None

__all__ = [
    "Vt100Parser",
]


# Regex matching any CPR response
# (Note that we use '\Z' instead of '$', because '$' could include a trailing
# newline.)
_cpr_response_re = re.compile("^" + re.escape("\x1b[") + r"\d+;\d+R\Z")

# Mouse events:
# Typical: "Esc[MaB*"  Urxvt: "Esc[96;14;13M" and for Xterm SGR: "Esc[<64;85;12M"
_mouse_event_re = re.compile("^" + re.escape("\x1b[") + r"(<?[\d;]+[mM]|M...)\Z")

# Regex matching any valid prefix of a CPR response.
# (Note that it doesn't contain the last character, the 'R'. The prefix has to
# be shorter.)
_cpr_response_prefix_re = re.compile("^" + re.escape("\x1b[") + r"[\d;]*\Z")

_mo
…[5433 chars omitted]…
eption:
            pass

    def _submit_secret_response(self, value: str) -> None:
        if not self._secret_state:
            return
        self._secret_state["response_queue"].put(value)
        self._secret_state = None
        self._secret_deadline = 0
        self._invalidate()

    def _cancel_secret_capture(self) -> None:
3191:    def _invalidate(self, min_interval: float = 0.25) -> None:

                image_bar,
                input_area,
                input_rule_bot,
                voice_status_bar,
                completions_menu,
            ] if item is not None
        ]

    def run(self):
        """Run the interactive CLI loop with persistent input at bottom."""
        # Detect light/dark terminal mode now (before pt grabs the tty).
        # Caches the result so subsequent _hex_to_ansi / style calls
        # don't risk re-querying mid-render.
        try:
    
…[8630 chars omitted]…
"3"), \
             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
            result = cli._confirm_destructive_slash(
                "reset",
                "This starts a fresh session.\nThe current conversation history will be discarded.",
            )

        # Choice "3" normalizes to "cancel", which returns None.
        assert result is None

        """Throttled UI repaint — prevents terminal blinking on slow/SSH connections."""
        if getattr(self, "_resize_recovery_pending", False):
            return
        now = time.monotonic()
        if hasattr(self, "_app") and self._app and (now - self._last_invalidate) >= min_interval:
            self._l
…[8629 chars omitted]…
            patch.object(cli, "_prompt_text_input", return_value="1"), \
             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
            result = cli._confirm_destructive_slash(
                "new",
                "This starts a fresh session.\nThe current conversation history will be discarded.",
            )

        assert result == "once"

                        _drop_path = _file_drop["path"]
                        _remainder = _file_
…[8629 chars omitted]…
"3"), \
             patch("cli.load_cli_config", return_value={"approvals": {"destructive_slash_confirm": True}}):
            result = cli._confirm_destructive_slash(
                "reset",
                "This starts a fresh session.\nThe current conversation history will be discarded.",
            )

        # Choice "3" normalizes to "cancel", which returns None.
        assert result is None

4. No-app path still uses the stdin fallback (existing behavior)
5. Empty choices returns None (existing behavior)
"""

import queue
import sys
import threading
import time
from unittest.mock import MagicMock, patch

import pytest


def _make_cli():
    """Minimal H
…[8630 chars omitted]…
           async def redraw_in_future() -> None:
                    await sleep(cast(float, self.min_redraw_interval) - diff)
                    schedule_redraw()

                self.loop.call_soon_threadsafe(
                    lambda: self.create_background_task(redraw_in_future())
                )
            else:
                schedule_redraw()
        else:
            schedule_redraw()

            if not getattr(_pt_renderer, "_hermes_osd_patched", False):
                def _patched_output_screen_diff(
                    app, output, screen, current_pos, color_depth,
                    previous_screen, last_style, is_done, full
…[8629 chars omitted]…
s (process_loop is
14181:        def process_loop():
14294:                        # AudioRecorder.start (lock acquire) never block process_loop —
14318:                    logger.warning("process_loop unhandled error (msg may be lost): %s", e)
14321:        process_thread = threading.Thread(target=process_loop, daemon=True)
14600:        # terminal modes — rather than from the background process_loop

                    except queue.Empty:
                        # Periodic config watcher — auto-reload MCP on mcp_servers change
                        if not self._agent_running:
                            self._check_config_mcp_changes()
                            # Check for background process notifications (completions
                            # and watch pattern matches) while agent is idle.
                            t
…[8066 chars omitted]…
before it can run one
            more poll iteration to clean up the subprocess, which was
            spawned with ``os.setsid`` and therefore survives as an orphan
            with PPID=1.

            Grace window (``HERMES_SIGTERM_GRACE``, default 1.5 s) gives
Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
AttributeError: type object 'Application' has no attribute 'loop'

        called. Return the value that was passed to
        :meth:`~prompt_toolkit.application.Application.exit`.

        This is the main entry point for a prompt_toolkit
        :class:`~prompt_toolkit.application.Application` and usually the only
        place where the event loop is actually running.

        :param pre_run: Optional callable, which is called right after the
            "reset" of the application.
        :param set_exception_handler: When set, 
…[6721 chars omitted]…
t as typeahead for next time.
                        store_typeahead(self.input, self.key_processor.empty_queue())

                return result

        @contextmanager
        def set_loop() -> Iterator[AbstractEventLoop]:
            loop = get_running_loop()
            self.loop = loop
            self._loop_thread = threading.current_thread()

            try:
                yield loop

        """Handle /update — update Hermes Agent to the latest version.

        In the classic CLI this exits the session and relaunches as
        ``hermes update`` so the user sees update output directly and gets
        the new version on next launch.

        Returns ``True`` when the update was confirmed (caller should trigger
        app exit so the relaunch is deferred to the main thread after
        prompt_toolkit cleans up terminal modes)
…[6778 chars omitted]…
ion.")
                print("   Re-enable via `approvals.mcp_reload_confirm: true` in config.yaml.")
            else:
                print("⚠️  Couldn't persist opt-out — reloading once.")

        with self._busy_command(self._slow_command_status(cmd_original)):
            self._reload_mcp()

    def _reload_mcp(self):
        """Reload MCP servers: disconnect all, re-read config.yaml, reconnect.

logger = logging.getLogger(__name__)


def _apply_bracketed_paste_timeout_patch():
    """Recover prompt_toolkit input after an incomplete bracketed paste.

    prompt_toolkit buffers all input after ``ESC[200~`` until it sees
    ``ESC[201~``.  If the end marker is lost, that normally traps every
    subsequent keystroke in the paste buffer.  Give the marker two seconds
    to arrive, then submit the buffered text as a paste and resume parsing
    the current input normally.

    Keep all imports local: this helper is also used by small standalone
    compatibility checks which execute its sour
…[3103 chars omitted]…

from hermes_cli.fallback_config import get_fallback_chain

# prompt_toolkit for fixed input area TUI
from prompt_toolkit.history import FileHistory
from prompt_toolkit.styles import Style as PTStyle
from prompt_toolkit.patch_stdout import patch_stdout
from prompt_toolkit.application import Application
from prompt_toolkit.layout import Layout, HSplit, Window, FormattedTextControl, ConditionalContainer

        """Run curses_single_select via run_in_terminal so prompt_toolkit handles terminal ownership cleanly."""
        import threading
        from hermes_cli.curses_ui import curses_
…[6086 chars omitted]…
nfo=True)
                return None

            # A real asyncio event loop returns a Handle when it queues work.
            # If an application-like object has no running callback mechanism,
            # don't wait forever for a modal which cannot be displayed.  The
 cli.py | 256 ++++++++++++++++++++++++++++++++++++++++++++++++-----------------
 1 file changed, 189 insertions(+), 67 deletions(-)

        assert result == "1"

    def test_non_main_thread_falls_back_to_stdin(self):
        """Off the main thread, _prompt_text_input_modal should use stdin fallback."""
        cli = _make_cli()
        result_holder = {}

        def run_on_daemon():
            # Patch platform to "linux" so the Windows check doesn't short-circuit.
            with patch.object
…[8629 chars omitted]…
ng = self._slash_confirm_deadline - _time.monotonic()
+                if remaining <= 0:
+                    break
                 try:
-                    result = response_queue.get(timeout=1)
-                    self._slash_confirm_state = None
-                    self._slash_confirm_deadline = 0
-                    self._restore_modal_input_snapshot()
-                    self._invalidate()

        """show_banner() no longer prints the activated skills line — it moved to run()."""
     
…[7489 chars omitted]…
======================== short test summary info ============================
FAILED tests/cli/test_cli_preloaded_skills.py::test_show_banner_does_not_print_skills
FAILED tests/cli/test_personality_none.py::TestGatewayPersonalityNone::test_none_clears_ephemeral_prompt
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 2 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
2 failed, 528 passed, 11 warnings in 214.84s (0:03:34)

+    prompt_toolkit buffers all input after ``ESC[200~`` until it sees
+    ``ESC[201~``.  If the end marker is lost, that normally traps every
+    subsequent keystroke in the paste buffer.  Give the marker two seconds
+    to arrive, then submit the buffered text as a paste and resume parsing
+    the current input normally.
+
+    Keep all imports local: this helper is also used by small standalone
+    compatibility checks which execute its source independently of cli.py.
+    """
+    import time as _time
+    import prompt_toolkit.input.vt100_parser as _vt100_parser
+   
…[8629 chars omitted]…
            result = response_queue.get(timeout=min(1.0, remaining))
                    break
                except queue.Empty:
                    now = _time.monotonic()
                    if now - last_countdown_refresh >= 5.0:
                        last_countdown_refresh = now
                        self._invalidate()
            return result
        finally:
            if in_main_thread:

+    """Recover prompt_toolkit input after an incomplete bracketed paste.
+
+    prompt_toolkit buffers all input after ``ESC[200~`` until it sees
+    ``ESC[201~``.  If the end marker is lost, that normally traps every
+    subsequent keystroke in the paste buffer.  Give the marker two seconds
+    to arrive, then submit the buffered text as a paste and resume parsing
+    the current input normally.
+
+    Keep all imports local: this helper is also used by small standalone
+    compatibility checks which execute its so
…[8629 chars omitted]…
              # cleanup rather than leaving the modal state stranded.
+                    logger.debug("Could not schedule slash confirmation cleanup", exc_info=True)
+                    _teardown_modal()
+                else:
+                    teardown_complete.wait()
 
     def _submit_slash_confirm_response(self, value: str | None) -> None:
         state = self._slash_confirm_state
 M cli.py

    python cli.py --list-tools             # List available tools and exit
"""

# IMPORTANT: hermes_bootstrap must be the very first import — UTF-8 stdio
# on Windows.  No-op on POSIX.  See hermes_bootstrap.py for full rationale.
try:
    import hermes_bootstrap  # noqa: F401
except ModuleNotFoundError:
    # Graceful fallback when hermes_bootstrap isn't registered in the venv
    # yet — happens during parti
…[8629 chars omitted]…
          return
        state["response_queue"].put(value)
        self._slash_confirm_state = None
        self._slash_confirm_deadline = 0
        self._invalidate()

    def _normalize_slash_confirm_choice(
        self,
        raw: str | None,
        choices: list[tuple[str, str, str]],
    ) -> str | None:
        if raw is None:
            return None
        choice_raw = raw.strip().lower()

 # on Windows.  No-op on POSIX.  See hermes_bootstrap.py for full rationale.
 try:
     import hermes_bootstrap  # noqa: F401
 except ModuleNotFoundError:
     # Graceful fallback when hermes_bootstrap isn't registered in the venv
     # yet — happens during partial ``hermes update`` where git-reset landed
     # new code but ``uv pip install -e .`` didn't finish.  Missing bootstrap
     # means UTF-8 stdio s
…[8629 chars omitted]…
ways",
             "always": "always",
             "remember": "always",
             "3": "cancel",
             "cancel": "cancel",
             "nevermind": "cancel",
             "no": "cancel",
             "n": "cancel",
         }
         allowed = {choice[0] for choice in choices}
         normalized = aliases.get(choice_raw)
         if normalized in allowed:
             return normalized

# which silently corrupts any non-ASCII file content.  We had three
# separate Windows sandbox regressions in one debug session before
# adding the explicit encoding.  This rule keeps new code honest.
select = ["PLW1514"]

[tool.ruff.lint.per-file-ignores]
# Tests can intentionally exercise locale-encoding edge cases.
"tests/**" = ["PLW1514"]
# Skills and plugins are partially user-authored — their own conventions.

+    except Exception:
+        # The CLI should remain importable if a future prompt_toolkit release
+        # changes its parser internals.  The compatibility failure is useful
+        # in logs, but must not prevent the interactive CLI from starting.
+        logger.debug("Could not install bracketed-paste recovery patch", exc_info=True)
+
+
+_apply_bracketed_paste_timeout_patch()
+
 # Suppress startup messages for clean CLI experience
 os.environ["HERMES_QUIET"] = "
…[8628 chars omitted]…
rt
+                    # cleanup rather than leaving the modal state stranded.
+                    logger.debug("Could not schedule slash confirmation cleanup", exc_info=True)
+                    _teardown_modal()
+                else:
+                    teardown_complete.wait()
 
     def _submit_slash_confirm_response(self, value: str | None) -> None:
         state = self._slash_confirm_state

        result = None
        try:
            if setup_error:
                logger.debug("Could not initialize slash confirmation modal: %s", setup_error[0])
                return None

            last_countdown_refresh = _time.monotonic()
            while True:
                # _submit_slash_confirm_response clears the shared UI deadline
                # before waking this worker; keep a local copy so that a fast
                # select
…[2592 chars omitted]…
te:
            return []

        title = state.get("title") or "Confirm action"
        detail = state.get("detail") or ""
        choices = state.get("choices") or []
        selected = state.get("selected", 0)

        def _panel_box_width(title_text: str, content_lines: list[str], min_width: int = 56, max_width: int = 86) -> int:
            term_cols = shutil.get_terminal_size((100, 20)).columns