Persistent bottom toolbar during command execution

CHANGELOG.md

@@ -1,10 +1,13 @@
-## 4.2.4 (TBD)
+## 4.3.0 (TBD)
 
 - Bug Fixes
     - Fixed the right prompt being redrawn beside every accepted command line in the scrollback.
       prompt-toolkit includes it in the final frame of each prompt, which is the frame left on the
       terminal; it is now hidden there, as the bottom toolbar already was, and stays on the live
       prompt line only
+- Enhancements
+    - `enable_bottom_toolbar=True` now keeps the toolbar visible and refreshing during command
+      execution
 
 ## 4.2.3 (September 2, 2026)
 

cmd2/cmd2.py

@@ -48,6 +48,7 @@
 from collections.abc import (
     Callable,
     Iterable,
+    Iterator,
     Mapping,
     Sequence,
 )
@@ -108,6 +109,7 @@
 from . import (
     argparse_completer,
     argparse_utils,
+    command_toolbar,
     constants,
     plugin,
     utils,
@@ -418,7 +420,7 @@ def __init__(
                              This allows CommandSets with custom constructor parameters to be
                              loaded.  This also allows the a set of CommandSets to be provided
                              when `auto_load_commands` is set to False
-        :param enable_bottom_toolbar: if ``True``, enables a bottom toolbar while at the main prompt.
+        :param enable_bottom_toolbar: if ``True``, enables a bottom toolbar at the main prompt and during commands.
                                       Override ``get_bottom_toolbar()`` to define its content.
         :param enable_rprompt: if ``True``, enables a right prompt while at the main prompt.
                                Override ``get_rprompt()`` to define its content.
@@ -561,6 +563,7 @@ def __init__(
         # custom prompt). Completion and UI logic should reference this variable
         # to ensure they modify the correct session state.
         self.active_session = self.main_session
+        self._command_toolbar: command_toolbar.CommandToolbar | None = None
 
         # Commands to exclude from the history command
         self.exclude_from_history = ["_eof", "history"]
@@ -1896,6 +1899,7 @@ def pfeedback(
                 rich_print_kwargs=rich_print_kwargs,
             )
 
+    @command_toolbar.suspend_toolbar
     def ppaged(
         self,
         *objects: Any,
@@ -2070,7 +2074,7 @@ def ppretty(
     def get_bottom_toolbar(self) -> AnyFormattedText:
         """Get the bottom toolbar content.
 
-        This method is called by prompt-toolkit while at the main prompt if ``enable_bottom_toolbar``
+        This method is called by prompt-toolkit at the main prompt and during commands if ``enable_bottom_toolbar``
         was set to ``True`` during initialization. Because prompt-toolkit executes this callback
         on every UI refresh (such as on every keypress or at scheduled refresh intervals), keeping
         this function highly optimized is critical to ensuring the CLI remains responsive.
@@ -2079,10 +2083,54 @@ def get_bottom_toolbar(self) -> AnyFormattedText:
         your application. This could be information like the application name, current state,
         or even a real-time clock.
 
+        During command execution this callback runs in a background UI thread. Protect shared
+        state with a lock when necessary. The toolbar is suspended while another prompt, pager,
+        or interactive shell owns the terminal.
+
         :return: Content to populate the bottom toolbar.
         """
         return None
 
+    @contextlib.contextmanager
+    def suspend_bottom_toolbar(self) -> Iterator[None]:
+        """Temporarily hide the command toolbar and give exclusive access to the terminal.
+
+        Use this context manager around application-specific calls to ``input()``, other
+        terminal UIs, or subprocesses that inherit the terminal. cmd2 automatically suspends
+        its toolbar for its own input prompts, pagers, and shell commands.
+        """
+        if self._command_toolbar is None:
+            yield
+        else:
+            with self._command_toolbar.suspend():
+                yield
+
+    @contextlib.contextmanager
+    def _command_toolbar_context(self) -> Iterator[None]:
+        """Display the toolbar around commands launched by the interactive command loop."""
+        if (
+            self._command_toolbar is not None
+            or self.main_session.bottom_toolbar is None
+            or not self._is_tty_session(self.main_session)
+        ):
+            yield
+            return
+
+        toolbar = command_toolbar.CommandToolbar(self)
+        try:
+            with self.sigint_protection:
+                toolbar.start()
+                self._command_toolbar = toolbar
+            yield
+        finally:
+            with self.sigint_protection:
+                try:
+                    toolbar.stop()
+                finally:
+                    # Always forget a toolbar that has been torn down. Keeping a failed
+                    # one would disable the toolbar for the rest of the session.
+                    self._command_toolbar = None
+
     def get_rprompt(self) -> AnyFormattedText:
         """Provide text to populate the prompt-toolkit right prompt.
 
@@ -3107,6 +3155,7 @@ def onecmd_plus_hooks(
 
         return stop
 
+    @command_toolbar.suspend_toolbar
     def _run_cmdfinalization_hooks(self, stop: bool, statement: Statement | None) -> bool:
         """Run the command finalization hooks."""
         if self._initial_termios_settings is not None and self.stdin.isatty():  # type: ignore[unreachable]
@@ -3342,32 +3391,50 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                 if shell:
                     kwargs["executable"] = shell
 
-            # For any stream that is a StdSim, we will use a pipe so we can capture its output
-            proc = subprocess.Popen(  # noqa: S602
-                statement.redirect_to,
-                stdin=subproc_stdin,
-                stdout=subprocess.PIPE if isinstance(self.stdout, utils.StdSim) else self.stdout,  # type: ignore[unreachable]
-                stderr=subprocess.PIPE if isinstance(sys.stderr, utils.StdSim) else sys.stderr,
-                shell=True,
-                **kwargs,
+            # Hand the pipe process the real terminal when there is one to inherit, since it
+            # may be interactive. Otherwise capture its output so it can pass through a Python
+            # stream, including the toolbar proxy which prints above the running display.
+            pipe_stdout = (
+                None
+                if isinstance(self.stdout, utils.StdSim)  # type: ignore[unreachable]
+                else command_toolbar.pipe_target(self.stdout)
             )
+            pipe_stderr = None if isinstance(sys.stderr, utils.StdSim) else command_toolbar.pipe_target(sys.stderr)
+
+            with contextlib.ExitStack() as terminal_stack:
+                # The toolbar can neither draw nor hold the keyboard while a pipe process owns
+                # the terminal, so step aside until that process has finished.
+                if pipe_stdout is not None or pipe_stderr is not None:
+                    terminal_stack.enter_context(self.suspend_bottom_toolbar())
+
+                proc = subprocess.Popen(  # noqa: S602
+                    statement.redirect_to,
+                    stdin=subproc_stdin,
+                    stdout=subprocess.PIPE if pipe_stdout is None else pipe_stdout,
+                    stderr=subprocess.PIPE if pipe_stderr is None else pipe_stderr,
+                    shell=True,
+                    **kwargs,
+                )
 
-            # Popen was called with shell=True so the user can chain pipe commands and redirect their output
-            # like: !ls -l | grep user | wc -l > out.txt. But this makes it difficult to know if the pipe process
-            # started OK, since the shell itself always starts. Therefore, we will wait a short time and check
-            # if the pipe process is still running.
-            with contextlib.suppress(subprocess.TimeoutExpired):
-                proc.wait(0.2)
+                # Popen was called with shell=True so the user can chain pipe commands and redirect their output
+                # like: !ls -l | grep user | wc -l > out.txt. But this makes it difficult to know if the pipe process
+                # started OK, since the shell itself always starts. Therefore, we will wait a short time and check
+                # if the pipe process is still running.
+                with contextlib.suppress(subprocess.TimeoutExpired):
+                    proc.wait(0.2)
+
+                # Check if the pipe process already exited
+                if proc.returncode is not None:
+                    subproc_stdin.close()
+                    new_stdout.close()
+                    raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run")
+                redir_saved_state.redirecting = True
+                cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr)
 
-            # Check if the pipe process already exited
-            if proc.returncode is not None:
-                subproc_stdin.close()
-                new_stdout.close()
-                raise RedirectionError(f"Pipe process exited with code {proc.returncode} before command could run")
-            redir_saved_state.redirecting = True
-            cmd_pipe_proc_reader = utils.ProcReader(proc, self.stdout, sys.stderr)
+                self.stdout = new_stdout
 
-            self.stdout = new_stdout
+                # Hold the suspension open until _restore_output() reaps the pipe process.
+                redir_saved_state.toolbar_suspension = terminal_stack.pop_all()
 
         elif statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND):
             if statement.redirect_to:
@@ -3419,29 +3486,35 @@ def _restore_output(self, statement: Statement, saved_redir_state: utils.Redirec
         :param statement: Statement object which contains the parsed input from the user
         :param saved_redir_state: contains information needed to restore state data
         """
-        if saved_redir_state.redirecting:
-            # If we redirected output to the clipboard
-            if (
-                statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND)
-                and not statement.redirect_to
-            ):
-                self.stdout.seek(0)
-                write_to_paste_buffer(self.stdout.read())
+        # The toolbar gets the terminal back once the pipe process is done with it.
+        with contextlib.ExitStack() as terminal_stack:
+            if saved_redir_state.toolbar_suspension is not None:
+                terminal_stack.callback(saved_redir_state.toolbar_suspension.close)
+                saved_redir_state.toolbar_suspension = None
+
+            if saved_redir_state.redirecting:
+                # If we redirected output to the clipboard
+                if (
+                    statement.redirector in (constants.REDIRECTION_OVERWRITE, constants.REDIRECTION_APPEND)
+                    and not statement.redirect_to
+                ):
+                    self.stdout.seek(0)
+                    write_to_paste_buffer(self.stdout.read())
 
-            with contextlib.suppress(BrokenPipeError):
-                # Close the file or pipe that stdout was redirected to
-                self.stdout.close()
+                with contextlib.suppress(BrokenPipeError):
+                    # Close the file or pipe that stdout was redirected to
+                    self.stdout.close()
 
-            # Restore self.stdout
-            self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout)
+                # Restore self.stdout
+                self.stdout = cast(TextIO, saved_redir_state.saved_self_stdout)
 
-            # Check if we need to wait for the process being piped to
-            if self._cur_pipe_proc_reader is not None:
-                self._cur_pipe_proc_reader.wait()
+                # Check if we need to wait for the process being piped to
+                if self._cur_pipe_proc_reader is not None:
+                    self._cur_pipe_proc_reader.wait()
 
-        # These are restored regardless of whether the command redirected
-        self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader
-        self._redirecting = saved_redir_state.saved_redirecting
+            # These are restored regardless of whether the command redirected
+            self._cur_pipe_proc_reader = saved_redir_state.saved_pipe_proc_reader
+            self._redirecting = saved_redir_state.saved_redirecting
 
     def get_command_func(self, command: str) -> BoundCommandFunc[...] | None:
         """Get the bound command function for a command.
@@ -3543,6 +3616,7 @@ def _is_tty_session(session: PromptSession[str]) -> bool:
         # a DummyOutput.
         return not isinstance(session.input, DummyInput)
 
+    @command_toolbar.suspend_toolbar
     def _read_raw_input(
         self,
         prompt: Callable[[], ANSI | str] | ANSI | str,
@@ -3824,7 +3898,11 @@ def _cmdloop(self) -> None:
         """
         try:
             # Run startup commands
-            stop = self.runcmds_plus_hooks(self._startup_commands)
+            if self._startup_commands:
+                with self._command_toolbar_context():
+                    stop = self.runcmds_plus_hooks(self._startup_commands)
+            else:
+                stop = False
             self._startup_commands.clear()
 
             while not stop:
@@ -3838,7 +3916,8 @@ def _cmdloop(self) -> None:
                     line = "_eof"
 
                 # Run the command along with all associated pre and post hooks
-                stop = self.onecmd_plus_hooks(line)
+                with self._command_toolbar_context():
+                    stop = self.onecmd_plus_hooks(line)
         finally:
             with self.sigint_protection:
                 # Shut down the alert thread.
@@ -4668,6 +4747,7 @@ def do_quit(self, _: argparse.Namespace) -> bool | None:
         self.last_result = True
         return True
 
+    @command_toolbar.suspend_toolbar
     def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], prompt: str = "Your choice? ") -> Any:
         """Present a menu to the user.
 
@@ -4876,6 +4956,7 @@ def _build_shell_parser(cls) -> Cmd2ArgumentParser:
 
     # Preserve quotes since we are passing these strings to the shell
     @with_argparser(_build_shell_parser, preserve_quotes=True)
+    @command_toolbar.suspend_toolbar
     def do_shell(self, args: argparse.Namespace) -> None:
         """Execute a command as if at the OS prompt."""
         import signal
@@ -4994,6 +5075,7 @@ def _restore_cmd2_env(self, cmd2_env: _SavedCmd2Env) -> None:
 
             readline.set_completer(cmd2_env.completer)
 
+    @command_toolbar.suspend_toolbar
     def _run_python(self, *, pyscript: str | None = None) -> bool | None:
         """Run an interactive Python shell or execute a pyscript file.
 
@@ -5205,6 +5287,7 @@ def _build_ipython_parser() -> Cmd2ArgumentParser:
         return argparse_utils.DEFAULT_ARGUMENT_PARSER(description="Run an interactive IPython shell.")
 
     @with_argparser(_build_ipython_parser)
+    @command_toolbar.suspend_toolbar
     def do_ipy(self, _: argparse.Namespace) -> bool | None:  # pragma: no cover
         """Run an interactive IPython shell.