Flush window title updates on the prompt event loop

Author: KennyMcSimpson

CHANGELOG.md

@@ -3,6 +3,9 @@
 - Bug Fixes
     - Fixed `@with_annotated` command `--help` output bypassing the application's output stream,
       leaving redirected help files empty and preventing help from being captured by embedded apps.
+    - Fixed `set_window_title()` leaving title changes buffered until the next prompt render.
+      Updates during a running prompt are sent through its event loop to avoid concurrent writes
+      from background threads.
 
 ## 4.3.0 (October 13, 2026)
 

cmd2/cmd2.py

@@ -71,7 +71,7 @@
     filters,
     print_formatted_text,
 )
-from prompt_toolkit.application import create_app_session, get_app
+from prompt_toolkit.application import create_app_session, get_app, get_app_or_none, get_app_session
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import Completer, DummyCompleter
 from prompt_toolkit.formatted_text import ANSI, AnyFormattedText
@@ -84,7 +84,7 @@
 from prompt_toolkit.layout.containers import ConditionalContainer, FloatContainer
 from prompt_toolkit.output import DummyOutput, 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,
@@ -5862,12 +5862,35 @@ def add_alert(
             self._alert_condition.notify_all()
 
     @staticmethod
-    def set_window_title(title: str) -> None:  # pragma: no cover
+    def set_window_title(title: str) -> None:
         """Set the terminal window title.
 
+        When a prompt-toolkit application is running, update the title on its event loop
+        to avoid writing to its output buffer from a background thread.
+        Calls use the current prompt-toolkit app session. Worker threads serving a custom
+        app session must inherit that session's context.
+
         :param title: the new window title
         """
-        set_title(title)
+        output = get_app_session().output
+
+        def set_title_and_flush() -> None:
+            output.set_title(title)
+            output.flush()
+
+        app = get_app_or_none()
+        loop = app.loop if app is not None else None
+        if loop is not None and loop.is_running():
+            try:
+                loop.call_soon_threadsafe(set_title_and_flush)
+            except RuntimeError:
+                # The application can finish and close its loop between the check and scheduling.
+                if not loop.is_closed():
+                    raise
+            else:
+                return
+
+        set_title_and_flush()
 
     def enable_command(self, command: str) -> None:
         """Enable a command by restoring its functions.

docs/features/prompt.md

@@ -54,6 +54,13 @@ Alerts can interact with the CLI in two ways:
 
 - [cmd2.Cmd.set_window_title][]
 
+Title changes are flushed without waiting for the next prompt. When a prompt-toolkit application is
+running, the update is scheduled on its event loop so background threads can change the title
+without writing to the prompt's output buffer directly.
+
+Calls use the current prompt-toolkit app session. If you create a custom session, pass its context
+to worker threads, for example with `contextvars.copy_context().run`.
+
 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.

tests/test_cmd2.py

@@ -1,6 +1,8 @@
 """Cmd2 unit/functional testing"""
 
+import asyncio
 import contextlib
+import contextvars
 import io
 import os
 import signal
@@ -17,11 +19,14 @@
 from unittest import mock
 
 import pytest
-from prompt_toolkit.application import create_app_session, get_app
+from prompt_toolkit.application import Application, create_app_session, get_app
+from prompt_toolkit.application.current import set_app
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import DummyCompleter
+from prompt_toolkit.data_structures import Size
 from prompt_toolkit.input import DummyInput, create_pipe_input
 from prompt_toolkit.output import DummyOutput
+from prompt_toolkit.output.vt100 import Vt100_Output
 from prompt_toolkit.shortcuts import CompleteStyle, PromptSession
 from rich.style import Style
 from rich.text import Text
@@ -4965,6 +4970,153 @@ def get_rprompt(self):
     assert True not in drawn_while_done, "the right prompt was drawn into the committed final frame"
 
 
+def test_set_window_title_is_flushed_during_a_command() -> None:
+    stream = io.StringIO()
+    output = Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    observed = []
+
+    class TitleApp(cmd2.Cmd):
+        def do_work(self, _statement):
+            self.set_window_title("Working…")
+            observed.append(stream.getvalue())
+
+    with create_app_session(input=DummyInput(), output=output):
+        app = TitleApp(allow_cli_args=False)
+        app.onecmd_plus_hooks("work")
+
+    assert observed == ["\x1b]2;Working…\x07"]
+
+
+@pytest.mark.parametrize("from_worker", [False, True])
+def test_set_window_title_while_a_prompt_is_running(from_worker) -> None:
+    stream = io.StringIO()
+    title_threads = []
+    observed = []
+    workers = []
+    loop_thread = threading.get_ident()
+
+    class TitleOutput(Vt100_Output):
+        def set_title(self, title):
+            title_threads.append(threading.get_ident())
+            super().set_title(title)
+
+    output = TitleOutput(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    with create_pipe_input() as pipe_input, create_app_session(input=pipe_input, output=output):
+        app = Application(input=pipe_input, output=output)
+        sent = False
+
+        def check_title():
+            # Observe before exiting, because the final render also flushes output.
+            observed.append("\x1b]2;Working…\x07" in stream.getvalue())
+            app.exit()
+
+        def update_title():
+            cmd2.Cmd.set_window_title("Working…")
+            app.loop.call_soon_threadsafe(check_title)
+
+        def after_render(_sender):
+            nonlocal sent
+            if sent:
+                return
+            sent = True
+            # Start after the first frame, so its flush cannot hide a buffered title.
+            if from_worker:
+                context = contextvars.copy_context()
+                worker = threading.Thread(target=context.run, args=(update_title,), daemon=True)
+                workers.append(worker)
+                worker.start()
+            else:
+                update_title()
+
+        app.after_render += after_render
+        app.run(pre_run=lambda: app.loop.call_later(5, app.exit), set_exception_handler=False)
+
+    for worker in workers:
+        worker.join(timeout=5)
+        assert not worker.is_alive()
+
+    assert observed == [True]
+    assert title_threads == [loop_thread]
+
+
+def test_set_window_title_without_a_terminal() -> None:
+    with create_app_session(input=DummyInput(), output=DummyOutput()):
+        cmd2.Cmd.set_window_title("Working…")
+
+
+def test_set_window_title_after_the_application_loop_has_closed() -> None:
+    stream = io.StringIO()
+    output = Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    app = Application(input=DummyInput(), output=output)
+    output.flush()
+    stream.seek(0)
+    stream.truncate()
+    loop = asyncio.new_event_loop()
+    loop.close()
+    app.loop = loop
+
+    with (
+        create_app_session(input=DummyInput(), output=output),
+        set_app(app),
+        mock.patch.object(output, "flush", wraps=output.flush) as flush,
+    ):
+        cmd2.Cmd.set_window_title("Finished")
+        flush.assert_called_once_with()
+
+    assert stream.getvalue() == "\x1b]2;Finished\x07"
+
+
+@pytest.mark.parametrize("close_during_scheduling", [True, False])
+def test_set_window_title_when_scheduling_fails(monkeypatch, close_during_scheduling) -> None:
+    stream = io.StringIO()
+    output = Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    app = Application(input=DummyInput(), output=output)
+    output.flush()
+    stream.seek(0)
+    stream.truncate()
+    loop = asyncio.new_event_loop()
+    app.loop = loop
+    schedule = loop.call_soon_threadsafe
+    ready = threading.Event()
+    loop.call_soon(ready.set)
+    worker = threading.Thread(target=loop.run_forever, daemon=True)
+
+    def fail_to_schedule(callback):
+        if not close_during_scheduling:
+            raise RuntimeError("scheduling failed")
+        # Close the real loop after set_window_title has observed it running.
+        schedule(loop.stop)
+        worker.join(timeout=5)
+        assert not worker.is_alive()
+        loop.close()
+        schedule(callback)
+
+    worker.start()
+    try:
+        assert ready.wait(timeout=5)
+        monkeypatch.setattr(loop, "call_soon_threadsafe", fail_to_schedule)
+        with (
+            create_app_session(input=DummyInput(), output=output),
+            set_app(app),
+            mock.patch.object(output, "flush", wraps=output.flush) as flush,
+        ):
+            if close_during_scheduling:
+                cmd2.Cmd.set_window_title("Finished")
+                flush.assert_called_once_with()
+                assert stream.getvalue() == "\x1b]2;Finished\x07"
+            else:
+                with pytest.raises(RuntimeError, match="scheduling failed"):
+                    cmd2.Cmd.set_window_title("Finished")
+                flush.assert_not_called()
+                assert stream.getvalue() == ""
+    finally:
+        if loop.is_running():
+            schedule(loop.stop)
+        worker.join(timeout=5)
+        assert not worker.is_alive()
+        loop.close()
+
+
 def test_get_bottom_toolbar(base_app: cmd2.Cmd) -> None:
     assert base_app.get_bottom_toolbar() is None