Flush window title updates without waiting for a prompt

CHANGELOG.md

@@ -1,6 +1,8 @@
 ## 4.4.0 (TBD)
 
 - Breaking Changes
+    - `set_window_title()` is now an instance method. Call `app.set_window_title(title)` or
+      `self.set_window_title(title)` so title changes use that application's terminal.
     - `History.get()` and `History.span()` now raise `HistoryIndexError`, an `IndexError` subclass,
       for invalid indices and bounded ranges selecting no history items. Unbounded ranges (`:` and
       `..`) can still return an empty dictionary. Starts before the beginning of history are
@@ -32,6 +34,9 @@
       it directly would then have retained current-session entries hidden from unbounded ranges and
       searches. Interactive `history` use was unaffected, since `cmd2` only truncates when
       persisting history at exit.
+    - Fixed `set_window_title()` leaving title changes buffered until the next prompt render. Title
+      writes and flushes are serialized with rendering on the application's output driver, including
+      during prompt setup and teardown. Output errors propagate to the calling thread.
 
 ## 4.3.1 (October 24, 2026)
 

cmd2/cmd2.py

@@ -82,9 +82,9 @@
 from prompt_toolkit.keys import Keys
 from prompt_toolkit.layout import Window, walk
 from prompt_toolkit.layout.containers import ConditionalContainer, FloatContainer
-from prompt_toolkit.output import DummyOutput, create_output
+from prompt_toolkit.output import DummyOutput, Output, create_output
 from prompt_toolkit.patch_stdout import patch_stdout
-from prompt_toolkit.shortcuts import CompleteStyle, PromptSession, choice, set_title
+from prompt_toolkit.shortcuts import CompleteStyle, PromptSession, choice
 from prompt_toolkit.styles import DynamicStyle
 from rich.console import (
     Group,
@@ -220,6 +220,37 @@ def __init__(self, msg: str = "") -> None:
     from prompt_toolkit.buffer import Buffer
 
 
+class _LockedOutput:
+    """Serialize prompt rendering and synchronous title changes on one output driver."""
+
+    def __init__(self, output: Output) -> None:
+        """Wrap the output without changing its terminal or rendering behavior."""
+        self._output = output
+        self._lock = threading.RLock()
+
+    def __getattr__(self, name: str) -> Any:
+        """Forward output properties and serialize every output method."""
+        attribute = getattr(self._output, name)
+        if not callable(attribute):
+            return attribute
+
+        def locked_call(*args: Any, **kwargs: Any) -> Any:
+            with self._lock:
+                return attribute(*args, **kwargs)
+
+        return locked_call
+
+    def set_title_and_flush(self, title: str) -> None:
+        """Keep the title write and flush together, including during prompt setup and teardown."""
+        with self._lock:
+            self._output.set_title(title)
+            self._output.flush()
+
+
+# Like prompt-toolkit's Windows10_Output, this proxy implements Output by delegation.
+Output.register(_LockedOutput)
+
+
 class _SavedCmd2Env:
     """cmd2 environment settings that are backed up when entering an interactive Python shell."""
 
@@ -794,6 +825,20 @@ def _handle_bracketed_paste(event: KeyPressEvent) -> None:
 
         return key_bindings
 
+    @property
+    def main_session(self) -> PromptSession[str]:
+        """The application's main prompt session."""
+        return self._main_session
+
+    @main_session.setter
+    def main_session(self, session: PromptSession[str]) -> None:
+        output = session.output
+        if isinstance(output, Output) and type(output) is not DummyOutput and not isinstance(output, _LockedOutput):
+            locked_output = cast(Output, _LockedOutput(output))
+            session.app.output = locked_output
+            session.app.renderer.output = locked_output
+        self._main_session = session
+
     def _create_main_session(
         self,
         *,
@@ -834,8 +879,8 @@ def _create_main_session(
             try:
                 if self.stdin != sys.stdin:
                     kwargs["input"] = create_input(stdin=self.stdin)
-                if self.stdout != sys.stdout:
-                    kwargs["output"] = create_output(stdout=self.stdout)
+                # Each Cmd owns its driver, so another app session cannot share its render buffer.
+                kwargs["output"] = create_output(stdout=self.stdout)
                 return PromptSession(**kwargs)
 
             except (NoConsoleScreenBufferError, AttributeError, ValueError):
@@ -5883,13 +5928,23 @@ def add_alert(
             self._alert_queue.append(alert)
             self._alert_condition.notify_all()
 
-    @staticmethod
-    def set_window_title(title: str) -> None:  # pragma: no cover
-        """Set the terminal window title.
+    def set_window_title(self, title: str) -> None:
+        """Set and immediately flush the title of this application's terminal.
+
+        Calls from worker threads use the same output as the main prompt, custom input prompts,
+        and select() menus, without requiring an inherited prompt-toolkit context.
+        The output driver serializes title changes with rendering, including while a prompt starts
+        or finishes. Output errors propagate to the caller instead of an event-loop callback.
 
         :param title: the new window title
         """
-        set_title(title)
+        output = self.main_session.output
+        if isinstance(output, _LockedOutput):
+            output.set_title_and_flush(title)
+        else:
+            # Non-interactive sessions use a DummyOutput, which needs no synchronization.
+            output.set_title(title)
+            output.flush()
 
     def enable_command(self, command: str) -> None:
         """Enable a command by restoring its functions.

docs/examples/embedding_in_textual.md

@@ -96,9 +96,6 @@ The example handles several details which are easy to miss when building somethi
 - The postparsing hooks stop running as soon as one of them sets `stop` or raises an exception, but
   the finalization hooks still run. If your application has other postparsing hooks, register the
   one which counts commands first so that it is never skipped.
-- `prompt-toolkit` buffers its output, so flush it after setting the window title when a command
-  starts. Otherwise the "running" state only reaches the Textual application after the command is
-  done.
 - Any program run in the terminal, for example by the `shell` command, can set the window title.
   Validate a status document before using it. Besides `json.JSONDecodeError`, `json.loads` raises
   `ValueError` for integers with too many digits and `RecursionError` for documents which are nested

docs/features/prompt.md

@@ -54,6 +54,15 @@ Alerts can interact with the CLI in two ways:
 
 - [cmd2.Cmd.set_window_title][]
 
+Call this method on your application instance, for example `app.set_window_title("Working…")`. The
+title is sent to the same terminal as that instance's main prompt, custom input prompts, and
+`select()` menus, including while a command is running.
+
+Title changes are flushed before the method returns. The output driver serializes title writes with
+prompt rendering, so background threads can call the method without inheriting a prompt-toolkit
+session context. Output errors, such as a closed terminal, propagate to the caller. Non-interactive
+sessions with a dummy output do not emit a title.
+
 The easiest way to understand these functions is to see the
 [async_printing.py](https://github.com/python-cmd2/cmd2/blob/main/examples/async_printing.py)
 example for a demonstration.

examples/textual_tty_embed.py

@@ -40,8 +40,6 @@
 from datetime import datetime
 from typing import Any, ClassVar
 
-from prompt_toolkit.application import get_app_session
-
 import cmd2
 
 # Prefix which marks a window title as a status update from the cmd2 application
@@ -92,8 +90,6 @@ def _report_status(self, state: str) -> None:
         # hook reports the end of the command. Writing to it then fails, which doesn't matter because nobody can see it.
         with contextlib.suppress(OSError):
             self.set_window_title(STATUS_PREFIX + json.dumps(status))
-            # prompt-toolkit buffers its output, so flush it or the "running" state would only appear after the command ends
-            get_app_session().output.flush()
 
     def preloop(self) -> None:
         """Report the initial status before the first prompt."""

tests/test_cmd2.py

@@ -4842,6 +4842,8 @@ def test_create_main_session_exception(monkeypatch):
     valid_session_mock = mock.MagicMock(spec=PromptSession)
     mock_session = mock.MagicMock(side_effect=[ValueError, valid_session_mock])
     monkeypatch.setattr("cmd2.cmd2.PromptSession", mock_session)
+    # Avoid requiring a physical console when isatty is mocked.
+    monkeypatch.setattr("cmd2.cmd2.create_output", lambda **_kwargs: DummyOutput())
 
     # Mock isatty to ensure we enter the try block
     with (
@@ -4995,6 +4997,8 @@ def test_create_main_session_no_console_error(monkeypatch):
     valid_session_mock = mock.MagicMock(spec=PromptSession)
     mock_session = mock.MagicMock(side_effect=[NoConsoleScreenBufferError, valid_session_mock])
     monkeypatch.setattr("cmd2.cmd2.PromptSession", mock_session)
+    # Avoid requiring a physical console when isatty is mocked.
+    monkeypatch.setattr("cmd2.cmd2.create_output", lambda **_kwargs: DummyOutput())
 
     # Mock isatty to ensure we enter the try block
     with (