fix(terminal): filter terminal query sequences from captured output

openhands-tools/openhands/tools/terminal/terminal/terminal_session.py

@@ -26,6 +26,7 @@
     escape_bash_special_chars,
     split_bash_commands,
 )
+from openhands.tools.terminal.utils.escape_filter import filter_terminal_queries
 
 
 logger = get_logger(__name__)
@@ -120,7 +121,12 @@ def _get_command_output(
         metadata: CmdOutputMetadata,
         continue_prefix: str = "",
     ) -> str:
-        """Get the command output with the previous command output removed."""
+        """Get the command output with the previous command output removed.
+
+        Also filters terminal query sequences that could cause visible escape
+        code garbage when the output is displayed.
+        See: https://github.com/OpenHands/software-agent-sdk/issues/2244
+        """
         # remove the previous command output from the new output if any
         if self.prev_output:
             command_output = raw_command_output.removeprefix(self.prev_output)
@@ -129,6 +135,11 @@ def _get_command_output(
             command_output = raw_command_output
         self.prev_output = raw_command_output  # update current command output anyway
         command_output = _remove_command_prefix(command_output, command)
+
+        # Filter terminal query sequences that would cause the terminal to
+        # respond when displayed, producing visible garbage
+        command_output = filter_terminal_queries(command_output)
+
         return command_output.rstrip()
 
     def _handle_completed_command(

openhands-tools/openhands/tools/terminal/utils/__init__.py

@@ -0,0 +1,14 @@
+"""Terminal tool utilities."""
+
+from openhands.tools.terminal.utils.command import (
+    escape_bash_special_chars,
+    split_bash_commands,
+)
+from openhands.tools.terminal.utils.escape_filter import filter_terminal_queries
+
+
+__all__ = [
+    "escape_bash_special_chars",
+    "split_bash_commands",
+    "filter_terminal_queries",
+]

openhands-tools/openhands/tools/terminal/utils/escape_filter.py

@@ -0,0 +1,83 @@
+"""Filter terminal query sequences from captured output.
+
+When CLI tools (like `gh`, `npm`, etc.) run inside a PTY, they may send
+terminal query sequences as part of their progress/spinner UI. These queries
+get captured as output. When displayed, the terminal processes them and
+responds, causing visible escape code garbage.
+
+This module provides filtering to remove these query sequences while
+preserving legitimate formatting escape codes (colors, bold, etc.).
+
+See: https://github.com/OpenHands/software-agent-sdk/issues/2244
+"""
+
+import re
+
+
+# Terminal query sequences that trigger responses (and cause visible garbage)
+# These should be stripped from captured output before display.
+#
+# Reference: ECMA-48, XTerm Control Sequences
+# https://invisible-island.net/xterm/ctlseqs/ctlseqs.html
+
+# DSR (Device Status Report) - cursor position query
+# Format: ESC [ 6 n  ->  Response: ESC [ row ; col R
+_DSR_PATTERN = re.compile(rb"\x1b\[6n")
+
+# OSC (Operating System Command) queries
+# Format: ESC ] Ps ; Pt (BEL | ST)
+# Common queries:
+#   OSC 10 ; ? - foreground color query
+#   OSC 11 ; ? - background color query
+#   OSC 4 ; index ; ? - palette color query
+# Terminators: BEL (\x07) or ST (ESC \)
+_OSC_QUERY_PATTERN = re.compile(
+    rb"\x1b\]"  # OSC introducer
+    rb"(?:10|11|4)"  # Color query codes (10=fg, 11=bg, 4=palette)
+    rb"[^"  # Match until terminator
+    rb"\x07\x1b]*"  # (not BEL or ESC)
+    rb"(?:\x07|\x1b\\)"  # BEL or ST terminator
+)
+
+# DA (Device Attributes) primary query
+# Format: ESC [ c  or  ESC [ 0 c
+_DA_PATTERN = re.compile(rb"\x1b\[0?c")
+
+# DA2 (Secondary Device Attributes) query
+# Format: ESC [ > c  or  ESC [ > 0 c
+_DA2_PATTERN = re.compile(rb"\x1b\[>0?c")
+
+# DECRQSS (Request Selection or Setting) - various terminal state queries
+# Format: ESC P $ q <setting> ST
+_DECRQSS_PATTERN = re.compile(
+    rb"\x1bP\$q"  # DCS introducer + DECRQSS
+    rb"[^\x1b]*"  # Setting identifier
+    rb"\x1b\\"  # ST terminator
+)
+
+
+def filter_terminal_queries(output: str) -> str:
+    """Filter terminal query sequences from captured terminal output.
+
+    Removes escape sequences that would cause the terminal to respond
+    when the output is displayed, while preserving legitimate formatting
+    sequences (colors, cursor movement, etc.).
+
+    Args:
+        output: Raw terminal output that may contain query sequences.
+
+    Returns:
+        Filtered output with query sequences removed.
+    """
+    # Convert to bytes for regex matching (escape sequences are byte-level)
+    output_bytes = output.encode("utf-8", errors="surrogateescape")
+
+    # Remove each type of query sequence
+    output_bytes = _DSR_PATTERN.sub(b"", output_bytes)
+    output_bytes = _OSC_QUERY_PATTERN.sub(b"", output_bytes)
+    output_bytes = _DA_PATTERN.sub(b"", output_bytes)
+    output_bytes = _DA2_PATTERN.sub(b"", output_bytes)
+    output_bytes = _DECRQSS_PATTERN.sub(b"", output_bytes)
+
+    # Convert back to string
+    return output_bytes.decode("utf-8", errors="surrogateescape")

tests/tools/terminal/test_escape_filter.py

@@ -0,0 +1,114 @@
+"""Tests for terminal escape sequence filtering.
+
+See: https://github.com/OpenHands/software-agent-sdk/issues/2244
+"""
+
+from openhands.tools.terminal.utils.escape_filter import filter_terminal_queries
+
+
+class TestFilterTerminalQueries:
+    """Tests for the filter_terminal_queries function."""
+
+    def test_dsr_query_removed(self):
+        """DSR (Device Status Report) queries should be removed."""
+        # \x1b[6n is the cursor position query
+        output = "some text\x1b[6nmore text"
+        result = filter_terminal_queries(output)
+        assert result == "some textmore text"
+
+    def test_osc_11_background_query_removed(self):
+        """OSC 11 (background color query) should be removed."""
+        # \x1b]11;?\x07 queries background color
+        output = "start\x1b]11;?\x07end"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+    def test_osc_10_foreground_query_removed(self):
+        """OSC 10 (foreground color query) should be removed."""
+        output = "start\x1b]10;?\x07end"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+    def test_osc_4_palette_query_removed(self):
+        """OSC 4 (palette color query) should be removed."""
+        output = "start\x1b]4;?\x07end"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+    def test_osc_with_st_terminator_removed(self):
+        """OSC queries with ST terminator should be removed."""
+        # ST terminator is \x1b\\
+        output = "start\x1b]11;?\x1b\\end"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+    def test_da_primary_query_removed(self):
+        """DA (Device Attributes) primary queries should be removed."""
+        # \x1b[c and \x1b[0c
+        output = "start\x1b[cend"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+        output2 = "start\x1b[0cend"
+        result2 = filter_terminal_queries(output2)
+        assert result2 == "startend"
+
+    def test_da2_secondary_query_removed(self):
+        """DA2 (Secondary Device Attributes) queries should be removed."""
+        # \x1b[>c and \x1b[>0c
+        output = "start\x1b[>cend"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+        output2 = "start\x1b[>0cend"
+        result2 = filter_terminal_queries(output2)
+        assert result2 == "startend"
+
+    def test_decrqss_query_removed(self):
+        """DECRQSS (Request Selection or Setting) queries should be removed."""
+        # \x1bP$q...\x1b\\
+        output = "start\x1bP$qsetting\x1b\\end"
+        result = filter_terminal_queries(output)
+        assert result == "startend"
+
+    def test_colors_preserved(self):
+        """ANSI color codes should NOT be removed."""
+        # Red text: \x1b[31m
+        output = "normal \x1b[31mred text\x1b[0m normal"
+        result = filter_terminal_queries(output)
+        assert result == output
+
+    def test_cursor_movement_preserved(self):
+        """Cursor movement codes should NOT be removed."""
+        # Move cursor: \x1b[H (home), \x1b[5A (up 5)
+        output = "start\x1b[Hmiddle\x1b[5Aend"
+        result = filter_terminal_queries(output)
+        assert result == output
+
+    def test_multiple_queries_removed(self):
+        """Multiple query sequences should all be removed."""
+        output = "\x1b[6n\x1b]11;?\x07text\x1b[6n"
+        result = filter_terminal_queries(output)
+        assert result == "text"
+
+    def test_mixed_queries_and_formatting(self):
+        """Queries removed while formatting preserved."""
+        # Color + query + more color
+        output = "\x1b[32mgreen\x1b[6nmore\x1b]11;?\x07text\x1b[0m"
+        result = filter_terminal_queries(output)
+        assert result == "\x1b[32mgreenmoretext\x1b[0m"
+
+    def test_empty_string(self):
+        """Empty string should return empty string."""
+        assert filter_terminal_queries("") == ""
+
+    def test_no_escape_sequences(self):
+        """Plain text without escape sequences passes through."""
+        output = "Hello, World!"
+        assert filter_terminal_queries(output) == output
+
+    def test_unicode_preserved(self):
+        """Unicode characters should be preserved."""
+        output = "Hello 🌍 World \x1b[6n with emoji"
+        result = filter_terminal_queries(output)
+        assert result == "Hello 🌍 World  with emoji"