diff --git a/code_puppy/agents/base_agent.py b/code_puppy/agents/base_agent.py index ac94d6e2e..d6e167a44 100644 --- a/code_puppy/agents/base_agent.py +++ b/code_puppy/agents/base_agent.py @@ -1772,6 +1772,16 @@ def _listen_for_ctrl_x_posix( data = stdin.read(1) if not data: break + # Drain multi-byte escape sequences (mouse events, arrow + # keys, etc.) so partial sequences don't leak into the + # stdin buffer for the next reader (e.g., prompt_toolkit). + if data == "\x1b": + from code_puppy.terminal_utils import ( + drain_stdin_escape_sequence, + ) + + drain_stdin_escape_sequence(stream=stdin) + continue if data == "\x18": # Ctrl+X try: on_escape() @@ -1787,6 +1797,16 @@ def _listen_for_ctrl_x_posix( except Exception: emit_warning("Cancel agent handler raised unexpectedly.") finally: + # Drain any remaining escape sequence bytes before restoring + # terminal attrs, so fragments don't leak to prompt_toolkit. + try: + from code_puppy.terminal_utils import ( + drain_stdin_escape_sequence, + ) + + drain_stdin_escape_sequence(stream=stdin) + except Exception: + pass termios.tcsetattr(fd, termios.TCSADRAIN, original_attrs) async def run_with_mcp( diff --git a/code_puppy/command_line/mcp/custom_server_form.py b/code_puppy/command_line/mcp/custom_server_form.py index 10e949aae..4a3f1fae2 100644 --- a/code_puppy/command_line/mcp/custom_server_form.py +++ b/code_puppy/command_line/mcp/custom_server_form.py @@ -611,7 +611,7 @@ def _(event): layout=layout, key_bindings=kb, full_screen=False, - mouse_support=True, + mouse_support=False, ) set_awaiting_user_input(True) @@ -634,10 +634,23 @@ def _(event): app.run(in_thread=True) finally: - # Exit alternate screen buffer - sys.stdout.write("\033[?1049l") - sys.stdout.flush() - set_awaiting_user_input(False) + # Explicitly disable mouse tracking as a safety net — if any + # previous prompt_toolkit Application left tracking enabled + # (e.g., due to an exception or threading race), this ensures + # the terminal stops sending mouse escape sequences. + try: + from code_puppy.terminal_utils import disable_mouse_tracking + + disable_mouse_tracking() + except Exception: + pass + finally: + # Exit alternate screen buffer — must always run even if + # disable_mouse_tracking fails, or the terminal stays in + # alt screen with input flags stuck. + sys.stdout.write("\033[?1049l") + sys.stdout.flush() + set_awaiting_user_input(False) # Clear exit message if not installing if self.result != "installed": diff --git a/code_puppy/terminal_utils.py b/code_puppy/terminal_utils.py index 6efb37e02..574a14b70 100644 --- a/code_puppy/terminal_utils.py +++ b/code_puppy/terminal_utils.py @@ -126,6 +126,67 @@ def reset_windows_terminal_full() -> None: flush_windows_keyboard_buffer() +def disable_mouse_tracking() -> None: + """Explicitly disable all mouse tracking escape sequences. + + Sends the standard VT100 escape sequences to disable mouse click, + any-event, urxvt, and SGR mouse tracking modes. This is a safety net + for when prompt_toolkit or other TUI components fail to clean up + mouse tracking on exit (e.g., due to exceptions or threading issues). + + Safe to call even if mouse tracking was never enabled. + """ + if platform.system() == "Windows": + return + + try: + sys.stdout.write( + "\x1b[?1000l" # Disable mouse click tracking + "\x1b[?1003l" # Disable any-event mouse tracking + "\x1b[?1015l" # Disable urxvt mouse mode + "\x1b[?1006l" # Disable SGR mouse mode + ) + sys.stdout.flush() + except Exception: + pass # Best effort — silently ignore errors + + +def drain_stdin_escape_sequence(stream=None, max_bytes: int = 256) -> None: + """Drain any pending multi-byte escape sequence bytes from stdin. + + When reading stdin byte-by-byte in cbreak/raw mode, mouse events and + other terminal escape sequences produce multi-byte sequences (e.g., + ``\\x1b[<0;15;20M`` for a mouse click). If a reader stops mid-sequence, + the remaining bytes stay in the stdin buffer and appear as garbled + characters when the next reader (e.g., prompt_toolkit) takes over. + + This function drains all currently pending bytes from stdin within a + short timeout window, preventing escape sequence fragments from leaking. + + Args: + stream: File-like object to drain. Defaults to ``sys.stdin``. + max_bytes: Safety cap to prevent infinite loops if stdin is flooded + (e.g., continuous mouse events). 256 bytes covers any realistic + burst of pending escape sequences. + """ + if platform.system() == "Windows": + return + + try: + import select + + fd = stream if stream is not None else sys.stdin + # Drain all pending bytes (10 ms timeout per byte — enough for + # escape sequences which arrive in a burst, but won't block on + # genuinely empty stdin). + drained = 0 + while drained < max_bytes and select.select([fd], [], [], 0.01)[0]: + fd.read(1) + drained += 1 + except Exception: + pass # Best effort — silently ignore errors + + def reset_unix_terminal() -> None: """Reset Unix/Linux/macOS terminal to sane state. diff --git a/code_puppy/tools/command_runner.py b/code_puppy/tools/command_runner.py index aa773acea..aec4fe5e6 100644 --- a/code_puppy/tools/command_runner.py +++ b/code_puppy/tools/command_runner.py @@ -392,6 +392,16 @@ def _listen_for_ctrl_x_posix( data = stdin.read(1) if not data: break + # Drain multi-byte escape sequences (mouse events, arrow + # keys, etc.) so partial sequences don't leak into the + # stdin buffer for the next reader (e.g., prompt_toolkit). + if data == "\x1b": + from code_puppy.terminal_utils import ( + drain_stdin_escape_sequence, + ) + + drain_stdin_escape_sequence(stream=stdin) + continue if data == "\x18": # Ctrl+X try: on_escape() @@ -400,6 +410,14 @@ def _listen_for_ctrl_x_posix( "Ctrl+X handler raised unexpectedly; Ctrl+C still works." ) finally: + # Drain any remaining escape sequence bytes before restoring + # terminal attrs, so fragments don't leak to prompt_toolkit. + try: + from code_puppy.terminal_utils import drain_stdin_escape_sequence + + drain_stdin_escape_sequence(stream=stdin) + except Exception: + pass termios.tcsetattr(fd, termios.TCSADRAIN, original_attrs) diff --git a/tests/agents/test_base_agent_full_coverage.py b/tests/agents/test_base_agent_full_coverage.py index 4165df12e..e33cfa07e 100644 --- a/tests/agents/test_base_agent_full_coverage.py +++ b/tests/agents/test_base_agent_full_coverage.py @@ -2129,9 +2129,11 @@ def test_loads_from_project_dir(self, agent, tmp_path): patch("code_puppy.config.CONFIG_DIR", str(tmp_path / "nonexistent")), patch( "pathlib.Path.exists", - side_effect=lambda self: str(self) == str(rules_file) - or str(self).endswith("AGENTS.md") - and "nonexistent" not in str(self), + side_effect=lambda self: ( + str(self) == str(rules_file) + or str(self).endswith("AGENTS.md") + and "nonexistent" not in str(self) + ), ), ): # Complex to test due to pathlib patching, just test cached path diff --git a/tests/command_line/test_model_settings_menu_coverage.py b/tests/command_line/test_model_settings_menu_coverage.py index 97e257f24..c96a224c8 100644 --- a/tests/command_line/test_model_settings_menu_coverage.py +++ b/tests/command_line/test_model_settings_menu_coverage.py @@ -151,10 +151,13 @@ def test_get_supported_settings(self, mock_supports): def test_load_model_settings_with_openai( self, mock_supports, mock_get_all, mock_effort, mock_verb ): - mock_supports.side_effect = lambda m, s: s in ( - "temperature", - "reasoning_effort", - "verbosity", + mock_supports.side_effect = lambda m, s: ( + s + in ( + "temperature", + "reasoning_effort", + "verbosity", + ) ) menu = _make_menu() menu._load_model_settings("gpt-5")