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()` and `Cmd.async_refresh_prompt()`.
 * 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:
@@ -5273,16 +5281,16 @@ class TestMyAppCase(Cmd2TestCase):
     def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:  # pragma: no cover
         """
         Display an important message to the user while they are at a command line prompt.
-        To the user it appears as if an alert message is printed above the prompt and their current input
-        text and cursor location is left alone.
+        To the user it appears as if an alert message is printed above the prompt and their
+        current input 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
-                   to guarantee the alert prints and to avoid raising a RuntimeError.
+        This function needs to acquire self.terminal_lock to ensure a prompt is on screen.
+        Therefore, it is best to acquire the lock before calling this function to avoid
+        raising a RuntimeError.
 
-                   This function is only needed when you need to print an alert while the main thread is blocking
-                   at the prompt. Therefore, this should never be called from the main thread. Doing so will
-                   raise a RuntimeError.
+        This function is only needed when you need to print an alert or update the prompt while the
+        main thread is blocking at the prompt. Therefore, this should never be called from the main
+        thread. Doing so will raise a RuntimeError.
 
         :param alert_msg: the message to display to the user
         :param new_prompt: If you also want to change the prompt that is displayed, then include it here.
@@ -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()
 
@@ -5346,30 +5349,50 @@ def async_alert(self, alert_msg: str, new_prompt: Optional[str] = None) -> None:
 
     def async_update_prompt(self, new_prompt: str) -> None:  # pragma: no cover
         """
-        Update the command line prompt while the user is still typing at it. This is good for alerting the user to
-        system changes dynamically in between commands. For instance you could alter the color of the prompt to
-        indicate a system status or increase a counter to report an event. If you do alter the actual text of the
-        prompt, it is best to keep the prompt the same width as what's on screen. Otherwise the user's input text will
-        be shifted and the update will not be seamless.
+        Update the command line prompt while the user is still typing at it.
 
-        IMPORTANT: This function will not update the prompt 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
-                   to guarantee the prompt changes and to avoid raising a RuntimeError.
+        This is good for alerting the user to system changes dynamically in between commands.
+        For instance you could alter the color of the prompt to indicate a system status or increase a
+        counter to report an event. If you do alter the actual text of the prompt, it is best to keep
+        the prompt the same width as what's on screen. Otherwise the user's input text will be shifted
+        and the update will not be seamless.
 
-                   This function is only needed when you need to update the prompt while the main thread is blocking
-                   at the prompt. Therefore, this should never be called from the main thread. Doing so will
-                   raise a RuntimeError.
-
-                   If user is at a continuation prompt while entering a multiline command, the onscreen prompt will
-                   not change. However, self.prompt will still be updated and display immediately after the multiline
-                   line command completes.
+        If user is at a continuation prompt while entering a multiline command, the onscreen prompt will
+        not change. However, self.prompt will still be updated and display immediately after the multiline
+        line command completes.
 
         :param new_prompt: what to change the prompt to
         :raises RuntimeError: if called from the main thread.
         :raises RuntimeError: if called while another thread holds `terminal_lock`
         """
         self.async_alert('', new_prompt)
 
+    def async_refresh_prompt(self) -> None:  # pragma: no cover
+        """
+        Refresh the oncreen prompt 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 this method is called. Call need_prompt_refresh() in an async print
+        thread to know when a refresh is needed.
+
+        :raises RuntimeError: if called from the main thread.
+        :raises RuntimeError: if called while another thread holds `terminal_lock`
+        """
+        self.async_alert('')
+
+    def need_prompt_refresh(self) -> bool:  # pragma: no cover
+        """Check whether the onscreen prompt needs to be asynchronously refreshed to match self.prompt."""
+        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

docs/features/prompt.rst

@@ -39,7 +39,7 @@ PythonScripting_ for an example of dynamically updating the prompt.
 Asynchronous Feedback
 ---------------------
 
-``cmd2`` provides two functions to provide asynchronous feedback to the user
+``cmd2`` provides these functions to provide asynchronous feedback to the user
 without interfering with the command line. This means the feedback is provided
 to the user when they are still entering text at the prompt. To use this
 functionality, the application must be running in a terminal that supports
@@ -52,6 +52,12 @@ all support these.
 .. automethod:: cmd2.Cmd.async_update_prompt
     :noindex:
 
+.. automethod:: cmd2.Cmd.async_refresh_prompt
+    :noindex:
+
+.. automethod:: cmd2.Cmd.need_prompt_refresh
+    :noindex:
+
 ``cmd2`` also provides a function to change the title of the terminal window.
 This feature requires the application be running in a terminal that supports
 VT100 control characters. Linux, Mac, and Windows 10 and greater all support
@@ -64,5 +70,3 @@ The easiest way to understand these functions is to see the AsyncPrinting_
 example for a demonstration.
 
 .. _AsyncPrinting: https://github.com/python-cmd2/cmd2/blob/master/examples/async_printing.py
-
-

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,10 +189,13 @@ 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:
+                # Otherwise check if the prompt needs to be updated or refreshed
+                elif self.prompt != new_prompt:
                     self.async_update_prompt(new_prompt)
 
+                elif self.need_prompt_refresh():
+                    self.async_refresh_prompt()
+
                 # Don't forget to release the lock
                 self.terminal_lock.release()