diff --git a/cli.py b/cli.py index 101ac997d80..2eef49b66d2 100644 --- a/cli.py +++ b/cli.py @@ -14702,6 +14702,22 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin): app.invalidate() # Refresh status line + # Post-turn terminal recovery (#33271): after an + # interrupt the prompt_toolkit renderer may have + # drifted from the physical terminal state — CSI 6n + # cursor position reports can leak as literal text + # (^[[19;1R), and the VT100 input parser can stall in + # a partial-escape state, accepting no further + # keystrokes. Drain stray escape bytes from the OS + # input buffer and force a clean renderer redraw. + if self._last_turn_interrupted: + try: + from hermes_cli.curses_ui import flush_stdin + flush_stdin() + except Exception: + pass + self._force_full_redraw() + # Goal continuation: if a standing goal is active, ask # the judge whether the turn satisfied it. If not, and # there's no real user message already queued, push the diff --git a/tests/cli/test_terminal_interrupt_recovery.py b/tests/cli/test_terminal_interrupt_recovery.py new file mode 100644 index 00000000000..b878e31b511 --- /dev/null +++ b/tests/cli/test_terminal_interrupt_recovery.py @@ -0,0 +1,127 @@ +"""Regression test for #33271: terminal recovery after interrupt. + +After an interrupt, the process_loop finally block must: + 1. Drain stray escape bytes from the OS input buffer (flush_stdin) + 2. Force a full prompt_toolkit renderer redraw + +Without this fix, CSI 6n cursor position reports can leak as literal +text (^[[19;1R) and the VT100 input parser can stall, accepting no +further keystrokes. +""" + +import pytest +from unittest.mock import MagicMock, patch, PropertyMock + + +@pytest.fixture +def cli(): + """Create a minimal HermesCLI mock with the required attributes.""" + from cli import HermesCLI + + instance = MagicMock(spec=HermesCLI) + instance._agent_running = True + instance._spinner_text = "thinking" + instance._tool_start_time = 1.0 + instance._pending_tool_info = {"name": "test"} + instance._last_scrollback_tool = "tool_a" + instance._last_turn_interrupted = False + instance._force_full_redraw = MagicMock() + instance._last_input_mode_recovery = 0.0 + instance._input_mode_recovery_notice_shown = False + + app = MagicMock() + app.invalidate = MagicMock() + instance._app = app + + return instance + + +class TestPostInterruptTerminalRecovery: + """Verify that the finally block in process_loop recovers the terminal + state after an interrupted agent turn.""" + + def test_no_recovery_when_turn_completes_normally(self, cli): + """_force_full_redraw should NOT be called when the turn finishes + normally (no interrupt).""" + cli._last_turn_interrupted = False + + # Simulate the finally block logic + if cli._last_turn_interrupted: + cli._force_full_redraw() + + cli._force_full_redraw.assert_not_called() + + def test_recovery_after_interrupt(self, cli): + """_force_full_redraw MUST be called when the turn was interrupted.""" + cli._last_turn_interrupted = True + + # Simulate the finally block logic + if cli._last_turn_interrupted: + try: + from hermes_cli.curses_ui import flush_stdin + flush_stdin() + except Exception: + pass + cli._force_full_redraw() + + cli._force_full_redraw.assert_called_once() + + @patch("hermes_cli.curses_ui.flush_stdin") + def test_flush_stdin_called_after_interrupt(self, mock_flush, cli): + """flush_stdin must be called to drain stray escape bytes.""" + cli._last_turn_interrupted = True + + if cli._last_turn_interrupted: + try: + from hermes_cli.curses_ui import flush_stdin + flush_stdin() + except Exception: + pass + cli._force_full_redraw() + + mock_flush.assert_called_once() + + @patch("hermes_cli.curses_ui.flush_stdin", side_effect=OSError("no tty")) + def test_flush_stdin_failure_does_not_prevent_redraw(self, mock_flush, cli): + """Even if flush_stdin fails (e.g., no TTY), _force_full_redraw must + still be called.""" + cli._last_turn_interrupted = True + + if cli._last_turn_interrupted: + try: + from hermes_cli.curses_ui import flush_stdin + flush_stdin() + except Exception: + pass + cli._force_full_redraw() + + cli._force_full_redraw.assert_called_once() + + def test_agent_running_cleared_on_normal_exit(self, cli): + """State flags must be reset regardless of interrupt status.""" + cli._last_turn_interrupted = False + cli._agent_running = True + cli._spinner_text = "active" + + # Simulate the finally block + cli._agent_running = False + cli._spinner_text = "" + + assert cli._agent_running is False + assert cli._spinner_text == "" + + def test_agent_running_cleared_on_interrupt(self, cli): + """State flags must be reset even after interrupt + recovery.""" + cli._last_turn_interrupted = True + cli._agent_running = True + cli._spinner_text = "active" + + # Simulate the finally block + cli._agent_running = False + cli._spinner_text = "" + if cli._last_turn_interrupted: + cli._force_full_redraw() + + assert cli._agent_running is False + assert cli._spinner_text == "" + cli._force_full_redraw.assert_called_once()