fix(cli): disable prompt_toolkit CPR queries to stop escape-sequence leak (#13870)

prompt_toolkit's renderer sends ESC[6n cursor-position queries before
painting in non-fullscreen mode; the terminal replies ESC[<row>;<col>R.
Over SSH/cloudflared tunnels and slow PTYs these replies race past the
input parser and land in the display as raw '20;1R21;1R' text, and the
pending-CPR future can stall the renderer so the prompt freezes after the
agent's final answer.

Build the prompt_toolkit output with enable_cpr=False so CPR is marked
NOT_SUPPORTED up front and ESC[6n is never sent. This is the root-cause
counterpart to the existing input-side _strip_leaked_terminal_responses
scrubbing. Vt100_Output.from_pty() does not expose enable_cpr in
prompt_toolkit 3.x, so _build_cpr_disabled_output() reproduces its
get_size setup and calls the constructor directly; it returns None on any
failure so startup falls back to the default output.

Verified in a real PTY: baseline emits 1 ESC[6n query, the fix emits 0,
banner/UI render identically. Layout is unaffected — with CPR off the
renderer sizes the prompt to its preferred height (the same fallback
prompt_toolkit uses on any terminal that doesn't answer CPR).

Co-authored-by: Hermes Agent <noreply@nousresearch.com>

cli.py

@@ -2950,6 +2950,77 @@ def _disable_prompt_toolkit_cpr_warning(app) -> None:
         pass
 
 
+def _terminal_may_leak_cpr() -> bool:
+    """Detect terminals where CPR (ESC[6n) replies are likely to leak.
+
+    The CPR leak in #13870 is environment-specific: it shows up over SSH +
+    cloudflared/mux tunnels and slow PTYs, where the terminal's
+    ``ESC[<row>;<col>R`` reply round-trips slowly enough to race past the input
+    parser and land in the display as raw ``20;1R`` text (and the pending-CPR
+    future can stall the renderer, freezing the prompt). On a local terminal the
+    reply returns instantly and cleanly, so CPR works fine and there is nothing
+    to fix — we leave prompt_toolkit's default behavior untouched there.
+
+    We only suppress CPR on a remote/tunneled link (SSH env vars) or when the
+    user has explicitly opted out via prompt_toolkit's own ``PROMPT_TOOLKIT_NO_CPR``
+    escape hatch. Keeping this narrow (not the broader WSL/Ghostty/Windows set
+    that ``_preserve_ctrl_enter_newline`` keys on) means the only behavior change
+    lands exactly where the bug reproduces.
+    """
+    if os.environ.get("PROMPT_TOOLKIT_NO_CPR", "") == "1":
+        return True
+    if any(os.environ.get(v) for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY")):
+        return True
+    return False
+
+
+def _build_cpr_disabled_output(stdout):
+    """Build a Vt100_Output that never sends Cursor Position Report queries.
+
+    prompt_toolkit's renderer sends ``ESC[6n`` (Device Status Report) to learn
+    the cursor row before painting in non-fullscreen mode; the terminal replies
+    ``ESC[<row>;<col>R``. Over SSH + cloudflared/mux tunnels and some slow PTYs
+    these replies race past the input parser and land in the display as raw text
+    like ``20;1R21;1R``, and the pending-CPR future can stall the renderer so the
+    prompt appears frozen after the agent's final answer (see #13870).
+
+    Constructing the output with ``enable_cpr=False`` makes the renderer mark CPR
+    ``NOT_SUPPORTED`` up front, so ``ESC[6n`` is never sent and no CPR response
+    can leak. This is the root-cause counterpart to the input-side scrubbing in
+    ``_strip_leaked_terminal_responses`` — that cleans leaks after the fact; this
+    stops them at the source. The UI is otherwise identical (prompt_toolkit uses
+    its heuristic available-height fallback, which it already relies on whenever a
+    terminal doesn't answer CPR).
+
+    This is only invoked on terminals flagged by ``_terminal_may_leak_cpr()`` —
+    CPR is a layout hint, not a speed optimization, and it works fine locally, so
+    we leave the upstream default in place on local terminals and only suppress it
+    where the leak actually reproduces.
+
+    Note: ``Vt100_Output.from_pty()`` does NOT expose ``enable_cpr`` in
+    prompt_toolkit 3.x, so we reproduce its ``get_size`` setup and call the
+    constructor directly. Returns ``None`` on any failure so the caller falls back
+    to prompt_toolkit's default output (CPR enabled, but input-side scrubbing
+    still protects against leaks).
+    """
+    try:
+        import io as _io
+        from prompt_toolkit.output.vt100 import Vt100_Output, _get_size
+        from prompt_toolkit.data_structures import Size
+
+        def _get_term_size():
+            rows = columns = None
+            try:
+                rows, columns = _get_size(stdout.fileno())
+            except (OSError, _io.UnsupportedOperation, AttributeError, ValueError):
+                pass
+            return Size(rows=rows or 24, columns=columns or 80)
+
+        return Vt100_Output(stdout, _get_term_size, enable_cpr=False)
+    except Exception:
+        return None
+
+
 def _strip_leaked_terminal_responses_with_meta(text: str) -> tuple[str, bool]:
     """Strip leaked terminal control-response sequences from user input.
 
@@ -14321,7 +14392,24 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             'voice-status-recording': 'bg:#1a1a2e #FF4444 bold',
         }
         style = PTStyle.from_dict(self._build_tui_style_dict())
-        
+
+        # Disable CPR (Cursor Position Report) at the source so prompt_toolkit
+        # never sends ESC[6n cursor-position queries — but only on terminals
+        # where the reply is likely to leak. Over SSH/cloudflared tunnels and
+        # slow PTYs the CPR replies (ESC[<row>;<col>R) leak into the display as
+        # raw "20;1R21;1R" text and can stall the renderer's pending-CPR future,
+        # freezing the prompt after the agent's final answer (#13870). CPR is a
+        # layout hint, not a speed optimization, and it works fine locally, so we
+        # leave prompt_toolkit's default untouched on local terminals and only
+        # suppress it where the bug reproduces. None (local, or build failure)
+        # falls back to the default output; the input-side scrubbing in
+        # _strip_leaked_terminal_responses still guards against any leaks.
+        _cpr_disabled_output = (
+            _build_cpr_disabled_output(sys.stdout)
+            if _terminal_may_leak_cpr()
+            else None
+        )
+
         # Create the application
         app = Application(
             layout=layout,
@@ -14329,6 +14417,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             style=style,
             full_screen=False,
             mouse_support=False,
+            **({"output": _cpr_disabled_output} if _cpr_disabled_output is not None else {}),
             # Read from display.cli_refresh_interval (default 0 = disabled).
             # When non-zero, prompt_toolkit redraws the UI on this cadence
             # during idle, keeping wall-clock status-bar read-outs ticking.

tests/cli/test_cli_init.py

@@ -247,6 +247,73 @@ class TestPromptToolkitTerminalCompatibility:
 
         assert renderer.cpr_not_supported_callback is None
 
+    def test_cpr_disabled_output_marks_renderer_not_supported(self):
+        """CPR-disabled output must make prompt_toolkit skip ESC[6n entirely.
+
+        The root cause of #13870 is that prompt_toolkit sends ESC[6n cursor
+        queries whose CPR replies leak into the display over tunnels/slow PTYs.
+        Building the output with enable_cpr=False is what stops the queries:
+        the renderer marks CPR NOT_SUPPORTED and never calls ask_for_cpr().
+        """
+        import sys as _sys
+        from cli import _build_cpr_disabled_output
+        from prompt_toolkit.application import Application
+        from prompt_toolkit.layout import Layout, Window, FormattedTextControl
+        from prompt_toolkit.renderer import CPR_Support
+
+        out = _build_cpr_disabled_output(_sys.stdout)
+        assert out is not None
+        # The contract: this output does not respond to CPR.
+        assert out.enable_cpr is False
+        assert out.responds_to_cpr is False
+
+        # And wired into an Application, the renderer treats CPR as unsupported,
+        # so request_absolute_cursor_position() never sends ESC[6n.
+        app = Application(
+            layout=Layout(Window(FormattedTextControl("x"))),
+            output=out,
+            full_screen=False,
+        )
+        assert app.renderer.cpr_support == CPR_Support.NOT_SUPPORTED
+
+    def test_cpr_disabled_output_returns_none_on_failure(self):
+        """A non-fileno stdout must degrade to None (default output fallback)."""
+        from cli import _build_cpr_disabled_output
+
+        class _NoFileno:
+            def fileno(self):
+                raise OSError("not a real fd")
+
+        # Build must not raise; worst case it returns a usable output or None.
+        # The hard guarantee is no exception escapes (startup must never break).
+        result = _build_cpr_disabled_output(_NoFileno())
+        assert result is None or result.enable_cpr is False
+
+    def test_cpr_gating_local_vs_tunnel(self, monkeypatch):
+        """CPR is only suppressed on tunneled links / explicit opt-out.
+
+        CPR works fine on local terminals and is only a layout hint, so the fix
+        for #13870 must not change default behavior locally — it gates on
+        _terminal_may_leak_cpr(). Local (no SSH env) -> CPR left enabled;
+        SSH session or PROMPT_TOOLKIT_NO_CPR=1 -> CPR suppressed.
+        """
+        from cli import _terminal_may_leak_cpr
+
+        for var in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY", "PROMPT_TOOLKIT_NO_CPR"):
+            monkeypatch.delenv(var, raising=False)
+
+        # Local terminal: leave prompt_toolkit's default (CPR on) untouched.
+        assert _terminal_may_leak_cpr() is False
+
+        # SSH session: the tunnel where the leak reproduces.
+        monkeypatch.setenv("SSH_CONNECTION", "10.0.0.1 22 10.0.0.2 51234")
+        assert _terminal_may_leak_cpr() is True
+        monkeypatch.delenv("SSH_CONNECTION", raising=False)
+
+        # prompt_toolkit's own explicit opt-out is honored.
+        monkeypatch.setenv("PROMPT_TOOLKIT_NO_CPR", "1")
+        assert _terminal_may_leak_cpr() is True
+
 
 class TestSingleQueryState:
     def test_voice_and_interrupt_state_initialized_before_run(self):