Fixed bug where async_alert() overwrites readline's incremental and non-incremental search prompts.

Author: kmvanbrunt

CHANGELOG.md

@@ -4,6 +4,10 @@
 * Bug Fixes
   * Fixed issue where persistent history file was not saved upon SIGHUP and SIGTERM signals.
   * Multiline commands are no longer fragmented in up-arrow history.
+  * Fixed bug where `async_alert()` overwrites readline's incremental and non-incremental search prompts.
+    * This fix introduces behavior where an updated prompt won't display after an aborted search
+      until a user presses Enter. See [async_printing.py](https://github.com/python-cmd2/cmd2/blob/master/examples/async_printing.py)
+      example for how to handle this case using `Cmd.need_prompt_refresh()`.
 * Enhancements
   * Removed dependency on `attrs` and replaced with [dataclasses](https://docs.python.org/3/library/dataclasses.html)
   * add `allow_clipboard` initialization parameter and attribute to disable ability to

cmd2/ansi.py

@@ -1058,7 +1058,7 @@ def async_alert_str(*, terminal_columns: int, prompt: str, line: str, cursor_off
     """Calculate the desired string, including ANSI escape codes, for displaying an asynchronous alert message.
 
     :param terminal_columns: terminal width (number of columns)
-    :param prompt: prompt that is displayed on the current line
+    :param prompt: current onscreen prompt
     :param line: current contents of the Readline line buffer
     :param cursor_offset: the offset of the current cursor position within line
     :param alert_msg: the message to display to the user
@@ -1071,9 +1071,9 @@ def async_alert_str(*, terminal_columns: int, prompt: str, line: str, cursor_off
     # Calculate how many terminal lines are taken up by all prompt lines except for the last one.
     # That will be included in the input lines calculations since that is where the cursor is.
     num_prompt_terminal_lines = 0
-    for line in prompt_lines[:-1]:
-        line_width = style_aware_wcswidth(line)
-        num_prompt_terminal_lines += int(line_width / terminal_columns) + 1
+    for prompt_line in prompt_lines[:-1]:
+        prompt_line_width = style_aware_wcswidth(prompt_line)
+        num_prompt_terminal_lines += int(prompt_line_width / terminal_columns) + 1
 
     # Now calculate how many terminal lines are take up by the input
     last_prompt_line = prompt_lines[-1]

cmd2/cmd2.py

@@ -130,8 +130,10 @@
 from .rl_utils import (
     RlType,
     rl_escape_prompt,
+    rl_get_display_prompt,
     rl_get_point,
     rl_get_prompt,
+    rl_in_search_mode,
     rl_set_prompt,
     rl_type,
     rl_warning,
@@ -3295,6 +3297,12 @@ def _set_up_cmd2_readline(self) -> _SavedReadlineSettings:
         """
         readline_settings = _SavedReadlineSettings()
 
+        if rl_type == RlType.GNU:
+            # To calculate line count when printing async_alerts, we rely on commands wider than
+            # the terminal to wrap across multiple lines. The default for horizontal-scroll-mode
+            # is "off" but a user may have overridden it in their readline initialization file.
+            readline.parse_and_bind("set horizontal-scroll-mode off")
+
         if self._completion_supported():
             # Set up readline for our tab completion needs
             if rl_type == RlType.GNU:
@@ -5277,7 +5285,7 @@ def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:
         text and cursor location is left alone.
 
         IMPORTANT: This function will not print an alert unless it can acquire self.terminal_lock to ensure
-                   a prompt is onscreen. Therefore, it is best to acquire the lock before calling this function
+                   a prompt is on screen. Therefore, it is best to acquire the lock before calling this function
                    to guarantee the alert prints and to avoid raising a RuntimeError.
 
                    This function is only needed when you need to print an alert while the main thread is blocking
@@ -5309,20 +5317,18 @@ def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:
             if new_prompt is not None:
                 self.prompt = new_prompt
 
-            # Check if the prompt to display has changed from what's currently displayed
-            cur_onscreen_prompt = rl_get_prompt()
-            new_onscreen_prompt = self.continuation_prompt if self._at_continuation_prompt else self.prompt
-
-            if new_onscreen_prompt != cur_onscreen_prompt:
+            # Check if the onscreen prompt needs to be refreshed to match self.prompt.
+            if self.need_prompt_refresh():
                 update_terminal = True
+                rl_set_prompt(self.prompt)
 
             if update_terminal:
                 import shutil
 
-                # Generate the string which will replace the current prompt and input lines with the alert
+                # Print a string which replaces the onscreen prompt and input lines with the alert.
                 terminal_str = ansi.async_alert_str(
                     terminal_columns=shutil.get_terminal_size().columns,
-                    prompt=cur_onscreen_prompt,
+                    prompt=rl_get_display_prompt(),
                     line=readline.get_line_buffer(),
                     cursor_offset=rl_get_point(),
                     alert_msg=alert_msg,
@@ -5333,9 +5339,6 @@ def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:
                 elif rl_type == RlType.PYREADLINE:
                     readline.rl.mode.console.write(terminal_str)
 
-                # Update Readline's prompt before we redraw it
-                rl_set_prompt(new_onscreen_prompt)
-
                 # Redraw the prompt and input lines below the alert
                 rl_force_redisplay()
 
@@ -5370,6 +5373,26 @@ def async_update_prompt(self, new_prompt: str) -> None:  # pragma: no cover
         """
         self.async_alert('', new_prompt)
 
+    def need_prompt_refresh(self) -> bool:  # pragma: no cover
+        """
+        Used by async print threads to check whether the onscreen prompt needs to be
+        refreshed to match self.prompt.
+
+        One case where the onscreen prompt and self.prompt can get out of sync is
+        when async_alert() is called while a user is in search mode (e.g. Ctrl-r).
+        To prevent overwriting readline's onscreen search prompt, self.prompt is updated
+        but readline's saved prompt isn't.
+
+        Therefore when a user aborts a search, the old prompt is still on screen until they
+        press Enter or async_alert() is called again. Use this function in an async print
+        thread to know when to make that extra call to async_alert().
+        """
+        if not (vt100_support and self.use_rawinput):
+            return False
+
+        # Don't overwrite a readline search prompt or a continuation prompt.
+        return not rl_in_search_mode() and not self._at_continuation_prompt and self.prompt != rl_get_prompt()
+
     @staticmethod
     def set_window_title(title: str) -> None:  # pragma: no cover
         """

cmd2/rl_utils.py

@@ -200,7 +200,7 @@ def rl_get_point() -> int:  # pragma: no cover
 
 
 def rl_get_prompt() -> str:  # pragma: no cover
-    """Gets Readline's current prompt"""
+    """Get Readline's prompt"""
     if rl_type == RlType.GNU:
         encoded_prompt = ctypes.c_char_p.in_dll(readline_lib, "rl_prompt").value
         if encoded_prompt is None:
@@ -221,6 +221,24 @@ def rl_get_prompt() -> str:  # pragma: no cover
     return rl_unescape_prompt(prompt)
 
 
+def rl_get_display_prompt() -> str:  # pragma: no cover
+    """
+    Get Readline's currently displayed prompt.
+
+    In GNU Readline, the displayed prompt sometimes differs from the prompt.
+    This occurs in functions that use the prompt string as a message area, such as incremental search.
+    """
+    if rl_type == RlType.GNU:
+        encoded_prompt = ctypes.c_char_p.in_dll(readline_lib, "rl_display_prompt").value
+        if encoded_prompt is None:
+            prompt = ''
+        else:
+            prompt = encoded_prompt.decode(encoding='utf-8')
+        return rl_unescape_prompt(prompt)
+    else:
+        return rl_get_prompt()
+
+
 def rl_set_prompt(prompt: str) -> None:  # pragma: no cover
     """
     Sets Readline's prompt
@@ -237,7 +255,8 @@ def rl_set_prompt(prompt: str) -> None:  # pragma: no cover
 
 
 def rl_escape_prompt(prompt: str) -> str:
-    """Overcome bug in GNU Readline in relation to calculation of prompt length in presence of ANSI escape codes
+    """
+    Overcome bug in GNU Readline in relation to calculation of prompt length in presence of ANSI escape codes
 
     :param prompt: original prompt
     :return: prompt safe to pass to GNU Readline
@@ -276,3 +295,32 @@ def rl_unescape_prompt(prompt: str) -> str:
         prompt = prompt.replace(escape_start, "").replace(escape_end, "")
 
     return prompt
+
+
+def rl_in_search_mode() -> bool:
+    """Check if readline is doing either an incremental (e.g. Ctrl-r) or non-incremental (e.g. Esc-p) search"""
+    if rl_type == RlType.GNU:
+        # GNU Readline defines constants that we can use to determine if in search mode.
+        #     RL_STATE_ISEARCH    0x0000080
+        #     RL_STATE_NSEARCH    0x0000100
+        IN_SEARCH_MODE = 0x0000180
+
+        readline_state = ctypes.c_int.in_dll(readline_lib, "rl_readline_state").value
+        return bool(IN_SEARCH_MODE & readline_state)
+    elif rl_type == RlType.PYREADLINE:
+        from pyreadline3.modes.emacs import (  # type: ignore[import]
+            EmacsMode,
+        )
+
+        # These search modes only apply to Emacs mode, which is the default.
+        if not isinstance(readline.rl.mode, EmacsMode):
+            return False
+
+        # While in search mode, the current keyevent function is set one of the following.
+        search_funcs = (
+            readline.rl.mode._process_incremental_search_keyevent,
+            readline.rl.mode._process_non_incremental_search_keyevent,
+        )
+        return readline.rl.mode.process_keyevent_queue[-1] in search_funcs
+    else:
+        return False

examples/async_printing.py

@@ -173,8 +173,8 @@ def _alerter_thread_func(self) -> None:
         self._next_alert_time = 0
 
         while not self._stop_event.is_set():
-            # Always acquire terminal_lock before printing alerts or updating the prompt
-            # To keep the app responsive, do not block on this call
+            # Always acquire terminal_lock before printing alerts or updating the prompt.
+            # To keep the app responsive, do not block on this call.
             if self.terminal_lock.acquire(blocking=False):
                 # Get any alerts that need to be printed
                 alert_str = self._generate_alert_str()
@@ -189,8 +189,9 @@ def _alerter_thread_func(self) -> None:
                     new_title = "Alerts Printed: {}".format(self._alert_count)
                     self.set_window_title(new_title)
 
-                # No alerts needed to be printed, check if the prompt changed
-                elif new_prompt != self.prompt:
+                # There are no alerts to print, but we should still check
+                # if the onscreen prompt needs to be updated.
+                elif self.prompt != new_prompt or self.need_prompt_refresh():
                     self.async_update_prompt(new_prompt)
 
                 # Don't forget to release the lock