From 2997f51f44553cd7dcee4c23438b8563f3052849 Mon Sep 17 00:00:00 2001
From: KennyMcSimpson <70093673+KennyMcSimpson@users.noreply.github.com>
Date: Sun, 25 Oct 2026 03:59:09 +0800
Subject: [PATCH 1/4] Flush window title updates on the prompt event loop

---
 CHANGELOG.md            |   3 +
 cmd2/cmd2.py            |  31 ++++++--
 docs/features/prompt.md |   7 ++
 tests/test_cmd2.py      | 154 +++++++++++++++++++++++++++++++++++++++-
 4 files changed, 190 insertions(+), 5 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 21fa3a869..11a87e29b 100644
--- a/CHANGELOG.md
+++ b/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)
 
diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index 94138cb32..c25fc347a 100644
--- a/cmd2/cmd2.py
+++ b/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.
diff --git a/docs/features/prompt.md b/docs/features/prompt.md
index 0f8a09082..92feee16a 100644
--- a/docs/features/prompt.md
+++ b/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.
diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py
index 2f39462d4..887d35374 100644
--- a/tests/test_cmd2.py
+++ b/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
 

From e49ec13bab68154eb0dc58b8430acf506283f80b Mon Sep 17 00:00:00 2001
From: KennyMcSimpson <70093673+KennyMcSimpson@users.noreply.github.com>
Date: Sun, 25 Oct 2026 09:27:55 +0800
Subject: [PATCH 2/4] Fix window title routing during select

---
 cmd2/cmd2.py            |  34 +++++--
 docs/features/prompt.md |   3 +-
 tests/test_cmd2.py      | 190 +++++++++++++++++++++++++++++++++++++++-
 3 files changed, 217 insertions(+), 10 deletions(-)

diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index c25fc347a..df1515f4a 100644
--- a/cmd2/cmd2.py
+++ b/cmd2/cmd2.py
@@ -71,7 +71,8 @@
     filters,
     print_formatted_text,
 )
-from prompt_toolkit.application import create_app_session, get_app, get_app_or_none, get_app_session
+from prompt_toolkit.application import create_app_session, get_app, get_app_session
+from prompt_toolkit.application.current import AppSession
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import Completer, DummyCompleter
 from prompt_toolkit.formatted_text import ANSI, AnyFormattedText
@@ -216,6 +217,11 @@ def __init__(self, msg: str = "") -> None:
     from prompt_toolkit.buffer import Buffer
 
 
+# select() creates a child session that ordinary background threads cannot see through contextvars.
+_select_app_sessions: dict[AppSession, list[AppSession]] = {}
+_select_app_sessions_lock = threading.Lock()
+
+
 class _SavedCmd2Env:
     """cmd2 environment settings that are backed up when entering an interactive Python shell."""
 
@@ -4788,10 +4794,20 @@ def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], p
                     fulloptions.append((opt[0], str(opt[0])))
 
         if self._is_tty_session(self.main_session):
+            parent_session = get_app_session()
             try:
                 while True:
-                    with create_app_session(input=self.main_session.input, output=self.main_session.output):
-                        result = choice(message=prompt, options=fulloptions)
+                    with create_app_session(input=self.main_session.input, output=self.main_session.output) as session:
+                        with _select_app_sessions_lock:
+                            sessions = _select_app_sessions.setdefault(parent_session, [])
+                            sessions.append(session)
+                        try:
+                            result = choice(message=prompt, options=fulloptions)
+                        finally:
+                            with _select_app_sessions_lock:
+                                sessions.remove(session)
+                                if not sessions:
+                                    del _select_app_sessions[parent_session]
                     if result is not None:
                         return result
             except KeyboardInterrupt:
@@ -5867,18 +5883,22 @@ def set_window_title(title: str) -> None:
 
         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.
+        Calls use the current prompt-toolkit app session, including any active select() menu.
+        Worker threads serving a custom app session must inherit that session's context.
 
         :param title: the new window title
         """
-        output = get_app_session().output
+        session = get_app_session()
+        with _select_app_sessions_lock:
+            while session in _select_app_sessions:
+                session = _select_app_sessions[session][-1]
+        app = session.app
+        output = app.output if app is not None else 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:
diff --git a/docs/features/prompt.md b/docs/features/prompt.md
index 9fca30810..c1dd1a2d9 100644
--- a/docs/features/prompt.md
+++ b/docs/features/prompt.md
@@ -56,7 +56,8 @@ Alerts can interact with the CLI in two ways:
 
 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.
+without writing to the prompt's output buffer directly. The title is sent to that application's
+output, including while a `select()` menu is active.
 
 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`.
diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py
index 887d35374..faf0c240b 100644
--- a/tests/test_cmd2.py
+++ b/tests/test_cmd2.py
@@ -19,7 +19,7 @@
 from unittest import mock
 
 import pytest
-from prompt_toolkit.application import Application, create_app_session, get_app
+from prompt_toolkit.application import Application, create_app_session, get_app, get_app_session
 from prompt_toolkit.application.current import set_app
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import DummyCompleter
@@ -4990,6 +4990,7 @@ def do_work(self, _statement):
 @pytest.mark.parametrize("from_worker", [False, True])
 def test_set_window_title_while_a_prompt_is_running(from_worker) -> None:
     stream = io.StringIO()
+    session_stream = io.StringIO()
     title_threads = []
     observed = []
     workers = []
@@ -5001,7 +5002,8 @@ def set_title(self, title):
             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):
+    session_output = Vt100_Output(session_stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    with create_pipe_input() as pipe_input, create_app_session(input=pipe_input, output=session_output):
         app = Application(input=pipe_input, output=output)
         sent = False
 
@@ -5037,6 +5039,190 @@ def after_render(_sender):
 
     assert observed == [True]
     assert title_threads == [loop_thread]
+    assert "\x1b]2;Working…\x07" not in session_stream.getvalue()
+
+
+@pytest.mark.parametrize("interrupt", [False, True])
+def test_set_window_title_from_a_worker_while_select_is_running(monkeypatch, interrupt) -> None:
+    stream = io.StringIO()
+    parent_stream = io.StringIO()
+    title_threads = []
+    worker_sessions = []
+    observed = []
+    rendered = threading.Event()
+    loop_thread = threading.get_ident()
+    parent = get_app_session()
+    run = Application.run
+
+    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)
+    parent_output = Vt100_Output(parent_stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+
+    def run_menu(menu, *args, **kwargs):
+        def check_title():
+            # Check before the menu's final render can flush a buffered title.
+            observed.append("\x1b]2;Working…\x07" in stream.getvalue())
+            if interrupt:
+                menu.exit(exception=KeyboardInterrupt)
+            else:
+                menu.exit(result="sweet")
+
+        def update_title():
+            worker_sessions.append(get_app_session())
+            if rendered.wait(timeout=5):
+                cmd2.Cmd.set_window_title("Working…")
+                menu.loop.call_soon_threadsafe(check_title)
+
+        # A plain worker sees the default session, not select's temporary one.
+        # An empty context keeps this true on Python builds that inherit context by default.
+        worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True)
+        menu.after_render += lambda _sender: rendered.set()
+        worker.start()
+        try:
+            return run(menu, *args, pre_run=lambda: menu.loop.call_later(5, lambda: menu.exit(result="timeout")), **kwargs)
+        finally:
+            worker.join(timeout=5)
+            assert not worker.is_alive()
+
+    with create_pipe_input() as pipe_input:
+        # Give the default session a safe output distinct from the menu's terminal.
+        monkeypatch.setattr(parent, "_input", pipe_input)
+        monkeypatch.setattr(parent, "_output", parent_output)
+        app = cmd2.Cmd(allow_cli_args=False, stdout=io.StringIO())
+        app.main_session = PromptSession(input=pipe_input, output=output)
+        monkeypatch.setattr(Application, "run", run_menu)
+
+        if interrupt:
+            with pytest.raises(KeyboardInterrupt):
+                app.select(["sweet", "salty"])
+        else:
+            assert app.select(["sweet", "salty"]) == "sweet"
+
+        assert observed == [True]
+        assert title_threads == [loop_thread]
+        assert worker_sessions == [parent]
+        assert "\x1b]2;Working…\x07" not in parent_stream.getvalue()
+
+        # Both normal return and interruption must restore the parent's routing.
+        cmd2.Cmd.set_window_title("Finished")
+        assert "\x1b]2;Finished\x07" in parent_stream.getvalue()
+        assert "\x1b]2;Finished\x07" not in stream.getvalue()
+
+
+@pytest.mark.parametrize("same_parent", [False, True])
+def test_set_window_title_during_nested_select_setup(monkeypatch, same_parent) -> None:
+    streams = [io.StringIO() for _ in range(4)]
+    outputs = [Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False) for stream in streams]
+    parent = get_app_session()
+    calls = 0
+
+    def update_title(title):
+        contextvars.Context().run(cmd2.Cmd.set_window_title, title)
+
+    def choose(**_kwargs):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            update_title("Outer before")
+            outer_session = app.main_session
+            app.main_session = PromptSession(input=pipe_input, output=outputs[2])
+            try:
+                if same_parent:
+                    # Re-enter the original session, temporarily replacing its existing mapping.
+                    contextvars.Context().run(app.select, ["inner"])
+                else:
+                    app.select(["inner"])
+            finally:
+                app.main_session = outer_session
+            update_title("Outer after")
+        else:
+            # choice has not constructed an Application yet: use the innermost session output.
+            update_title("Inner")
+            with create_app_session(input=DummyInput(), output=outputs[3]):
+                cmd2.Cmd.set_window_title("Unrelated")
+        return "selected"
+
+    with create_pipe_input() as pipe_input:
+        monkeypatch.setattr(parent, "_input", pipe_input)
+        monkeypatch.setattr(parent, "_output", outputs[0])
+        app = cmd2.Cmd(allow_cli_args=False)
+        app.main_session = PromptSession(input=pipe_input, output=outputs[1])
+        monkeypatch.setattr("cmd2.cmd2.choice", choose)
+        assert app.select(["outer"]) == "selected"
+        update_title("Finished")
+
+    assert calls == 2
+    assert streams[0].getvalue() == "\x1b]2;Finished\x07"
+    assert streams[1].getvalue().endswith("\x1b]2;Outer before\x07\x1b]2;Outer after\x07")
+    assert streams[2].getvalue().endswith("\x1b]2;Inner\x07")
+    assert streams[3].getvalue() == "\x1b]2;Unrelated\x07"
+
+
+def test_set_window_title_after_select_menus_finish_out_of_order(monkeypatch) -> None:
+    streams = [io.StringIO() for _ in range(3)]
+    outputs = [Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False) for stream in streams]
+    entered = [threading.Event(), threading.Event()]
+    released = [threading.Event(), threading.Event()]
+    results = []
+    errors = []
+
+    def choose(*, options, **_kwargs):
+        index = options[0][0]
+        entered[index].set()
+        assert released[index].wait(timeout=5)
+        return index
+
+    def select(app, index):
+        try:
+            results.append(app.select([(index, str(index))]))
+        # Preserve worker failures for the main test thread instead of relying on pytest's warning hook.
+        except BaseException as ex:  # noqa: BLE001
+            errors.append(ex)
+
+    with create_pipe_input() as pipe_input:
+        parent = get_app_session()
+        monkeypatch.setattr(parent, "_input", pipe_input)
+        monkeypatch.setattr(parent, "_output", outputs[0])
+        monkeypatch.setattr("cmd2.cmd2.choice", choose)
+        apps = [cmd2.Cmd(allow_cli_args=False) for _ in range(2)]
+        for index, app in enumerate(apps):
+            app.main_session = PromptSession(input=pipe_input, output=outputs[index + 1])
+        for output, stream in zip(outputs, streams, strict=True):
+            output.flush()
+            stream.seek(0)
+            stream.truncate()
+        workers = [
+            threading.Thread(target=contextvars.Context().run, args=(select, app, index), daemon=True)
+            for index, app in enumerate(apps)
+        ]
+        try:
+            workers[0].start()
+            assert entered[0].wait(timeout=5)
+            workers[1].start()
+            assert entered[1].wait(timeout=5)
+            cmd2.Cmd.set_window_title("Both active")
+            released[0].set()
+            workers[0].join(timeout=5)
+            assert not workers[0].is_alive()
+            cmd2.Cmd.set_window_title("Second active")
+        finally:
+            for event in released:
+                event.set()
+            for worker in workers:
+                if worker.ident is not None:
+                    worker.join(timeout=5)
+                    assert not worker.is_alive()
+        cmd2.Cmd.set_window_title("Finished")
+
+    assert not errors
+    assert results == [0, 1]
+    assert streams[0].getvalue() == "\x1b]2;Finished\x07"
+    assert streams[1].getvalue() == ""
+    assert streams[2].getvalue() == "\x1b]2;Both active\x07\x1b]2;Second active\x07"
 
 
 def test_set_window_title_without_a_terminal() -> None:

From c78994f8ec153d0475c28c8514d77d3fe298f3cc Mon Sep 17 00:00:00 2001
From: KennyMcSimpson <70093673+KennyMcSimpson@users.noreply.github.com>
Date: Sun, 25 Oct 2026 12:09:38 +0800
Subject: [PATCH 3/4] Explain select session traversal for title updates

---
 cmd2/cmd2.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index df1515f4a..db65ed6e1 100644
--- a/cmd2/cmd2.py
+++ b/cmd2/cmd2.py
@@ -5890,6 +5890,9 @@ def set_window_title(title: str) -> None:
         """
         session = get_app_session()
         with _select_app_sessions_lock:
+            # Workers may still see the parent session instead of select()'s temporary one.
+            # Follow registered children to the innermost active select() session.
+            # [-1] chooses the most recently registered active child when menus share a parent.
             while session in _select_app_sessions:
                 session = _select_app_sessions[session][-1]
         app = session.app

From a22715862f304a07db4059d1b4576d4ced59b350 Mon Sep 17 00:00:00 2001
From: KennyMcSimpson <70093673+KennyMcSimpson@users.noreply.github.com>
Date: Thu, 29 Oct 2026 03:49:56 +0800
Subject: [PATCH 4/4] Route synchronous title updates through each Cmd output

---
 CHANGELOG.md                          |   8 +-
 cmd2/cmd2.py                          | 111 ++++---
 docs/examples/embedding_in_textual.md |   3 -
 docs/features/prompt.md               |  15 +-
 examples/textual_tty_embed.py         |   4 -
 tests/test_cmd2.py                    | 344 +------------------
 tests/test_window_title.py            | 461 ++++++++++++++++++++++++++
 7 files changed, 539 insertions(+), 407 deletions(-)
 create mode 100644 tests/test_window_title.py

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2b3e127c5..0758e1de1 100644
--- a/CHANGELOG.md
+++ b/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
@@ -28,9 +30,9 @@
     - History regular expressions containing range separators are now handled as regex searches.
       Other selectors are treated as numbers or ranges only when they match the supported syntax;
       strings such as URLs, `a:b`, `+1`, and `1_0` remain literal searches.
-    - 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.
+    - 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)
 
diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index 773b6d6bb..a46d8cea2 100644
--- a/cmd2/cmd2.py
+++ b/cmd2/cmd2.py
@@ -71,8 +71,7 @@
     filters,
     print_formatted_text,
 )
-from prompt_toolkit.application import create_app_session, get_app, get_app_session
-from prompt_toolkit.application.current import AppSession
+from prompt_toolkit.application import create_app_session, get_app
 from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
 from prompt_toolkit.completion import Completer, DummyCompleter
 from prompt_toolkit.formatted_text import ANSI, AnyFormattedText
@@ -83,7 +82,7 @@
 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
 from prompt_toolkit.styles import DynamicStyle
@@ -221,9 +220,35 @@ def __init__(self, msg: str = "") -> None:
     from prompt_toolkit.buffer import Buffer
 
 
-# select() creates a child session that ordinary background threads cannot see through contextvars.
-_select_app_sessions: dict[AppSession, list[AppSession]] = {}
-_select_app_sessions_lock = threading.Lock()
+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:
@@ -800,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,
         *,
@@ -840,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):
@@ -4798,20 +4837,10 @@ def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], p
                     fulloptions.append((opt[0], str(opt[0])))
 
         if self._is_tty_session(self.main_session):
-            parent_session = get_app_session()
             try:
                 while True:
-                    with create_app_session(input=self.main_session.input, output=self.main_session.output) as session:
-                        with _select_app_sessions_lock:
-                            sessions = _select_app_sessions.setdefault(parent_session, [])
-                            sessions.append(session)
-                        try:
-                            result = choice(message=prompt, options=fulloptions)
-                        finally:
-                            with _select_app_sessions_lock:
-                                sessions.remove(session)
-                                if not sessions:
-                                    del _select_app_sessions[parent_session]
+                    with create_app_session(input=self.main_session.input, output=self.main_session.output):
+                        result = choice(message=prompt, options=fulloptions)
                     if result is not None:
                         return result
             except KeyboardInterrupt:
@@ -5899,44 +5928,24 @@ def add_alert(
             self._alert_queue.append(alert)
             self._alert_condition.notify_all()
 
-    @staticmethod
-    def set_window_title(title: str) -> None:
-        """Set the terminal window title.
+    def set_window_title(self, title: str) -> None:
+        """Set and immediately flush the title of this application's terminal.
 
-        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, including any active select() menu.
-        Worker threads serving a custom app session must inherit that session's context.
+        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
         """
-        session = get_app_session()
-        with _select_app_sessions_lock:
-            # Workers may still see the parent session instead of select()'s temporary one.
-            # Follow registered children to the innermost active select() session.
-            # [-1] chooses the most recently registered active child when menus share a parent.
-            while session in _select_app_sessions:
-                session = _select_app_sessions[session][-1]
-        app = session.app
-        output = app.output if app is not None else session.output
-
-        def set_title_and_flush() -> None:
+        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()
 
-        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.
 
diff --git a/docs/examples/embedding_in_textual.md b/docs/examples/embedding_in_textual.md
index e7373bbc6..d00e95807 100644
--- a/docs/examples/embedding_in_textual.md
+++ b/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
diff --git a/docs/features/prompt.md b/docs/features/prompt.md
index c1dd1a2d9..011517407 100644
--- a/docs/features/prompt.md
+++ b/docs/features/prompt.md
@@ -54,13 +54,14 @@ 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. The title is sent to that application's
-output, including while a `select()` menu is active.
-
-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`.
+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)
diff --git a/examples/textual_tty_embed.py b/examples/textual_tty_embed.py
index c93bdc5cb..2b306ab76 100755
--- a/examples/textual_tty_embed.py
+++ b/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."""
diff --git a/tests/test_cmd2.py b/tests/test_cmd2.py
index faf0c240b..487fbf7a8 100644
--- a/tests/test_cmd2.py
+++ b/tests/test_cmd2.py
@@ -1,8 +1,6 @@
 """Cmd2 unit/functional testing"""
 
-import asyncio
 import contextlib
-import contextvars
 import io
 import os
 import signal
@@ -19,14 +17,11 @@
 from unittest import mock
 
 import pytest
-from prompt_toolkit.application import Application, create_app_session, get_app, get_app_session
-from prompt_toolkit.application.current import set_app
+from prompt_toolkit.application import create_app_session, get_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
@@ -4847,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 (
@@ -4970,339 +4967,6 @@ 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()
-    session_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)
-    session_output = Vt100_Output(session_stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
-    with create_pipe_input() as pipe_input, create_app_session(input=pipe_input, output=session_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]
-    assert "\x1b]2;Working…\x07" not in session_stream.getvalue()
-
-
-@pytest.mark.parametrize("interrupt", [False, True])
-def test_set_window_title_from_a_worker_while_select_is_running(monkeypatch, interrupt) -> None:
-    stream = io.StringIO()
-    parent_stream = io.StringIO()
-    title_threads = []
-    worker_sessions = []
-    observed = []
-    rendered = threading.Event()
-    loop_thread = threading.get_ident()
-    parent = get_app_session()
-    run = Application.run
-
-    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)
-    parent_output = Vt100_Output(parent_stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
-
-    def run_menu(menu, *args, **kwargs):
-        def check_title():
-            # Check before the menu's final render can flush a buffered title.
-            observed.append("\x1b]2;Working…\x07" in stream.getvalue())
-            if interrupt:
-                menu.exit(exception=KeyboardInterrupt)
-            else:
-                menu.exit(result="sweet")
-
-        def update_title():
-            worker_sessions.append(get_app_session())
-            if rendered.wait(timeout=5):
-                cmd2.Cmd.set_window_title("Working…")
-                menu.loop.call_soon_threadsafe(check_title)
-
-        # A plain worker sees the default session, not select's temporary one.
-        # An empty context keeps this true on Python builds that inherit context by default.
-        worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True)
-        menu.after_render += lambda _sender: rendered.set()
-        worker.start()
-        try:
-            return run(menu, *args, pre_run=lambda: menu.loop.call_later(5, lambda: menu.exit(result="timeout")), **kwargs)
-        finally:
-            worker.join(timeout=5)
-            assert not worker.is_alive()
-
-    with create_pipe_input() as pipe_input:
-        # Give the default session a safe output distinct from the menu's terminal.
-        monkeypatch.setattr(parent, "_input", pipe_input)
-        monkeypatch.setattr(parent, "_output", parent_output)
-        app = cmd2.Cmd(allow_cli_args=False, stdout=io.StringIO())
-        app.main_session = PromptSession(input=pipe_input, output=output)
-        monkeypatch.setattr(Application, "run", run_menu)
-
-        if interrupt:
-            with pytest.raises(KeyboardInterrupt):
-                app.select(["sweet", "salty"])
-        else:
-            assert app.select(["sweet", "salty"]) == "sweet"
-
-        assert observed == [True]
-        assert title_threads == [loop_thread]
-        assert worker_sessions == [parent]
-        assert "\x1b]2;Working…\x07" not in parent_stream.getvalue()
-
-        # Both normal return and interruption must restore the parent's routing.
-        cmd2.Cmd.set_window_title("Finished")
-        assert "\x1b]2;Finished\x07" in parent_stream.getvalue()
-        assert "\x1b]2;Finished\x07" not in stream.getvalue()
-
-
-@pytest.mark.parametrize("same_parent", [False, True])
-def test_set_window_title_during_nested_select_setup(monkeypatch, same_parent) -> None:
-    streams = [io.StringIO() for _ in range(4)]
-    outputs = [Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False) for stream in streams]
-    parent = get_app_session()
-    calls = 0
-
-    def update_title(title):
-        contextvars.Context().run(cmd2.Cmd.set_window_title, title)
-
-    def choose(**_kwargs):
-        nonlocal calls
-        calls += 1
-        if calls == 1:
-            update_title("Outer before")
-            outer_session = app.main_session
-            app.main_session = PromptSession(input=pipe_input, output=outputs[2])
-            try:
-                if same_parent:
-                    # Re-enter the original session, temporarily replacing its existing mapping.
-                    contextvars.Context().run(app.select, ["inner"])
-                else:
-                    app.select(["inner"])
-            finally:
-                app.main_session = outer_session
-            update_title("Outer after")
-        else:
-            # choice has not constructed an Application yet: use the innermost session output.
-            update_title("Inner")
-            with create_app_session(input=DummyInput(), output=outputs[3]):
-                cmd2.Cmd.set_window_title("Unrelated")
-        return "selected"
-
-    with create_pipe_input() as pipe_input:
-        monkeypatch.setattr(parent, "_input", pipe_input)
-        monkeypatch.setattr(parent, "_output", outputs[0])
-        app = cmd2.Cmd(allow_cli_args=False)
-        app.main_session = PromptSession(input=pipe_input, output=outputs[1])
-        monkeypatch.setattr("cmd2.cmd2.choice", choose)
-        assert app.select(["outer"]) == "selected"
-        update_title("Finished")
-
-    assert calls == 2
-    assert streams[0].getvalue() == "\x1b]2;Finished\x07"
-    assert streams[1].getvalue().endswith("\x1b]2;Outer before\x07\x1b]2;Outer after\x07")
-    assert streams[2].getvalue().endswith("\x1b]2;Inner\x07")
-    assert streams[3].getvalue() == "\x1b]2;Unrelated\x07"
-
-
-def test_set_window_title_after_select_menus_finish_out_of_order(monkeypatch) -> None:
-    streams = [io.StringIO() for _ in range(3)]
-    outputs = [Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False) for stream in streams]
-    entered = [threading.Event(), threading.Event()]
-    released = [threading.Event(), threading.Event()]
-    results = []
-    errors = []
-
-    def choose(*, options, **_kwargs):
-        index = options[0][0]
-        entered[index].set()
-        assert released[index].wait(timeout=5)
-        return index
-
-    def select(app, index):
-        try:
-            results.append(app.select([(index, str(index))]))
-        # Preserve worker failures for the main test thread instead of relying on pytest's warning hook.
-        except BaseException as ex:  # noqa: BLE001
-            errors.append(ex)
-
-    with create_pipe_input() as pipe_input:
-        parent = get_app_session()
-        monkeypatch.setattr(parent, "_input", pipe_input)
-        monkeypatch.setattr(parent, "_output", outputs[0])
-        monkeypatch.setattr("cmd2.cmd2.choice", choose)
-        apps = [cmd2.Cmd(allow_cli_args=False) for _ in range(2)]
-        for index, app in enumerate(apps):
-            app.main_session = PromptSession(input=pipe_input, output=outputs[index + 1])
-        for output, stream in zip(outputs, streams, strict=True):
-            output.flush()
-            stream.seek(0)
-            stream.truncate()
-        workers = [
-            threading.Thread(target=contextvars.Context().run, args=(select, app, index), daemon=True)
-            for index, app in enumerate(apps)
-        ]
-        try:
-            workers[0].start()
-            assert entered[0].wait(timeout=5)
-            workers[1].start()
-            assert entered[1].wait(timeout=5)
-            cmd2.Cmd.set_window_title("Both active")
-            released[0].set()
-            workers[0].join(timeout=5)
-            assert not workers[0].is_alive()
-            cmd2.Cmd.set_window_title("Second active")
-        finally:
-            for event in released:
-                event.set()
-            for worker in workers:
-                if worker.ident is not None:
-                    worker.join(timeout=5)
-                    assert not worker.is_alive()
-        cmd2.Cmd.set_window_title("Finished")
-
-    assert not errors
-    assert results == [0, 1]
-    assert streams[0].getvalue() == "\x1b]2;Finished\x07"
-    assert streams[1].getvalue() == ""
-    assert streams[2].getvalue() == "\x1b]2;Both active\x07\x1b]2;Second active\x07"
-
-
-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
 
@@ -5333,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 (
diff --git a/tests/test_window_title.py b/tests/test_window_title.py
new file mode 100644
index 000000000..ddd257fb8
--- /dev/null
+++ b/tests/test_window_title.py
@@ -0,0 +1,461 @@
+"""Window title behavior through real prompt-toolkit output and applications."""
+
+import asyncio
+import contextvars
+import io
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import ExitStack
+
+import pytest
+from prompt_toolkit.application import Application, create_app_session
+from prompt_toolkit.application.current import set_app
+from prompt_toolkit.data_structures import Size
+from prompt_toolkit.input import DummyInput, create_pipe_input
+from prompt_toolkit.output import DummyOutput, Output
+from prompt_toolkit.output.vt100 import Vt100_Output
+from prompt_toolkit.shortcuts import PromptSession
+
+import cmd2
+
+
+class TtyStream(io.StringIO):
+    def isatty(self):
+        return True
+
+
+def make_app(monkeypatch, pipe_input, output):
+    # Use the real VT100 driver on every platform without requiring a physical console.
+    with monkeypatch.context() as terminal:
+        terminal.setattr("cmd2.cmd2.create_input", lambda **_kwargs: pipe_input)
+        terminal.setattr("cmd2.cmd2.create_output", lambda **_kwargs: output)
+        app = cmd2.Cmd(stdin=TtyStream(), stdout=output.stdout, allow_cli_args=False)
+    output.flush()
+    output.stdout.seek(0)
+    output.stdout.truncate()
+    return app
+
+
+def title_output(stream):
+    return Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+
+
+def test_title_during_a_command_uses_the_instance_terminal(monkeypatch):
+    stream = TtyStream()
+    foreign = TtyStream()
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+        app.do_work = lambda _statement: app.set_window_title("Working…")
+        app.onecmd_plus_hooks("work")
+        assert stream.getvalue() == "\x1b]2;Working…\x07"
+        assert foreign.getvalue() == ""
+        assert isinstance(app.main_session.output, Output)
+        assert app.main_session.output.stdout is stream
+        assert app.main_session.output.get_size() == Size(rows=24, columns=80)
+
+
+@pytest.mark.parametrize("from_worker", [False, True])
+@pytest.mark.parametrize("custom_prompt", [False, True])
+def test_title_before_during_and_after_a_real_prompt(monkeypatch, from_worker, custom_prompt):
+    stream = TtyStream()
+    foreign = TtyStream()
+    observed = []
+    errors = []
+    workers = []
+    run = Application.run
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+        app.set_window_title("Before")
+        sent = False
+
+        def update_title(prompt_app):
+            try:
+                app.set_window_title("During")
+                observed.append("\x1b]2;During\x07" in stream.getvalue())
+            except BaseException as error:  # noqa: BLE001
+                errors.append(error)
+            finally:
+                prompt_app.loop.call_soon_threadsafe(lambda: prompt_app.exit(result="done"))
+
+        def after_render(prompt_app):
+            nonlocal sent
+            if sent:
+                return
+            sent = True
+            if from_worker:
+                worker = threading.Thread(target=contextvars.Context().run, args=(update_title, prompt_app), daemon=True)
+                workers.append(worker)
+                worker.start()
+            else:
+                update_title(prompt_app)
+
+        def run_prompt(prompt_app, *args, **kwargs):
+            prompt_app.after_render += after_render
+            kwargs["set_exception_handler"] = False
+            return run(prompt_app, *args, **kwargs)
+
+        monkeypatch.setattr(Application, "run", run_prompt)
+        if custom_prompt:
+            result = app.read_input("custom> ")
+        else:
+            result = app._read_raw_input("prompt> ", app.main_session, set_exception_handler=False)
+        for worker in workers:
+            worker.join(timeout=5)
+            assert not worker.is_alive()
+        app.set_window_title("After")
+
+    assert result == "done"
+    assert not errors
+    assert observed == [True]
+    assert "\x1b]2;Before\x07" in stream.getvalue()
+    assert "\x1b]2;After\x07" in stream.getvalue()
+    assert foreign.getvalue() == ""
+
+
+@pytest.mark.parametrize("interrupt", [False, True])
+def test_worker_titles_during_a_real_select_and_after_cleanup(monkeypatch, interrupt):
+    stream = TtyStream()
+    foreign = TtyStream()
+    observed = []
+    errors = []
+    run = Application.run
+
+    def run_menu(menu, *args, **kwargs):
+        sent = False
+        workers = []
+
+        def update_title():
+            try:
+                app.set_window_title("Menu")
+                observed.append("\x1b]2;Menu\x07" in stream.getvalue())
+            except BaseException as error:  # noqa: BLE001
+                errors.append(error)
+            finally:
+                if interrupt:
+                    menu.loop.call_soon_threadsafe(lambda: menu.exit(exception=KeyboardInterrupt))
+                else:
+                    menu.loop.call_soon_threadsafe(lambda: menu.exit(result="sweet"))
+
+        def after_render(_sender):
+            nonlocal sent
+            if not sent:
+                sent = True
+                worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True)
+                workers.append(worker)
+                worker.start()
+
+        menu.after_render += after_render
+        try:
+            return run(menu, *args, set_exception_handler=False, **kwargs)
+        finally:
+            for worker in workers:
+                worker.join(timeout=5)
+                assert not worker.is_alive()
+
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+        monkeypatch.setattr(Application, "run", run_menu)
+        if interrupt:
+            with pytest.raises(KeyboardInterrupt):
+                app.select(["sweet", "salty"])
+        else:
+            assert app.select(["sweet", "salty"]) == "sweet"
+        app.set_window_title("Finished")
+
+    assert not errors
+    assert observed == [True]
+    assert "\x1b]2;Finished\x07" in stream.getvalue()
+    assert foreign.getvalue() == ""
+
+
+def test_instance_titles_do_not_cross_between_concurrent_select_menus(monkeypatch):
+    streams = [TtyStream(), TtyStream()]
+    rendered = [threading.Event(), threading.Event()]
+    menus = []
+    run = Application.run
+
+    def run_menu(menu, *args, **kwargs):
+        index = next(index for index, stream in enumerate(streams) if menu.output.stdout is stream)
+        menus.append(menu)
+        menu.after_render += lambda _sender: rendered[index].set()
+        return run(menu, *args, set_exception_handler=False, **kwargs)
+
+    monkeypatch.setattr(Application, "run", run_menu)
+    with ExitStack() as stack:
+        inputs = [stack.enter_context(create_pipe_input()) for _ in streams]
+        apps = [
+            make_app(monkeypatch, pipe_input, title_output(stream)) for pipe_input, stream in zip(inputs, streams, strict=True)
+        ]
+        with ThreadPoolExecutor(max_workers=2) as executor:
+            results = [executor.submit(app.select, ["selected"]) for app in apps]
+            try:
+                assert all(event.wait(timeout=5) for event in rendered)
+                apps[0].set_window_title("A")
+                apps[1].set_window_title("B")
+                inputs[0].send_text("\r")
+                assert results[0].result(timeout=5) == "selected"
+                apps[0].set_window_title("A finished")
+                apps[1].set_window_title("B active")
+            finally:
+                for menu in menus:
+                    loop = menu.loop
+                    if loop is not None and loop.is_running():
+                        loop.call_soon_threadsafe(lambda menu=menu: menu.exit(result="selected"))
+                for result in results:
+                    result.result(timeout=5)
+
+    assert "\x1b]2;A\x07" in streams[0].getvalue()
+    assert "\x1b]2;A finished\x07" in streams[0].getvalue()
+    assert "\x1b]2;B\x07" not in streams[0].getvalue()
+    assert "\x1b]2;B\x07" in streams[1].getvalue()
+    assert "\x1b]2;B active\x07" in streams[1].getvalue()
+    assert "\x1b]2;A\x07" not in streams[1].getvalue()
+
+
+def test_title_during_interrupted_select_setup_leaves_no_stale_routing(monkeypatch):
+    stream = TtyStream()
+    foreign = TtyStream()
+
+    def interrupt_before_application(**_kwargs):
+        app.set_window_title("Setting up")
+        raise KeyboardInterrupt
+
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+        monkeypatch.setattr("cmd2.cmd2.choice", interrupt_before_application)
+        with pytest.raises(KeyboardInterrupt):
+            app.select(["choice"])
+        app.set_window_title("After interrupt")
+
+    assert "\x1b]2;Setting up\x07" in stream.getvalue()
+    assert "\x1b]2;After interrupt\x07" in stream.getvalue()
+    assert foreign.getvalue() == ""
+
+
+@pytest.mark.parametrize("operation", ["set_title", "flush"])
+@pytest.mark.parametrize("from_worker", [False, True])
+def test_title_output_errors_reach_the_caller_during_a_prompt(monkeypatch, operation, from_worker):
+    stream = TtyStream()
+    caught = []
+    loop_errors = []
+    workers = []
+
+    class FailingOutput(Vt100_Output):
+        fail_flush = False
+
+        def set_title(self, title):
+            if operation == "set_title" and title == "Fail":
+                raise OSError("terminal closed")
+            super().set_title(title)
+            if title == "Fail":
+                self.fail_flush = operation == "flush"
+
+        def flush(self):
+            if self.fail_flush:
+                self.fail_flush = False
+                raise OSError("terminal closed")
+            super().flush()
+
+    output = FailingOutput(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+        app = make_app(monkeypatch, pipe_input, output)
+        sent = False
+
+        def update_title():
+            try:
+                app.set_window_title("Fail")
+            except OSError as error:
+                caught.append(str(error))
+            finally:
+                app.main_session.app.loop.call_soon_threadsafe(lambda: app.main_session.app.exit(result="done"))
+
+        def after_render(_sender):
+            nonlocal sent
+            if sent:
+                return
+            sent = True
+            app.main_session.app.loop.set_exception_handler(lambda _loop, context: loop_errors.append(context))
+            if from_worker:
+                worker = threading.Thread(target=contextvars.copy_context().run, args=(update_title,), daemon=True)
+                workers.append(worker)
+                worker.start()
+            else:
+                update_title()
+
+        app.main_session.app.after_render += after_render
+        assert app._read_raw_input("prompt> ", app.main_session, set_exception_handler=False) == "done"
+        for worker in workers:
+            worker.join(timeout=5)
+            assert not worker.is_alive()
+
+    assert caught == ["terminal closed"]
+    assert not loop_errors
+
+
+def test_title_is_not_queued_when_the_loop_stops_during_scheduling(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    loop = asyncio.new_event_loop()
+    ready = threading.Event()
+    loop.call_soon(ready.set)
+    schedule = loop.call_soon_threadsafe
+    runner = threading.Thread(target=loop.run_forever, daemon=True)
+
+    def stop_before_queuing(callback):
+        schedule(loop.stop)
+        runner.join(timeout=5)
+        assert not runner.is_alive()
+        schedule(callback)
+
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+        app = make_app(monkeypatch, pipe_input, output)
+        app.main_session.app.loop = loop
+        runner.start()
+        try:
+            assert ready.wait(timeout=5)
+            monkeypatch.setattr(loop, "call_soon_threadsafe", stop_before_queuing)
+            with set_app(app.main_session.app):
+                app.set_window_title("Finished")
+            assert stream.getvalue() == "\x1b]2;Finished\x07"
+        finally:
+            if loop.is_running():
+                schedule(loop.stop)
+            runner.join(timeout=5)
+            assert not runner.is_alive()
+            app.main_session.app.loop = None
+            loop.close()
+
+
+def test_render_flush_and_fallback_title_writes_are_serialized(monkeypatch):
+    stream = TtyStream()
+    entered = threading.Event()
+    release = threading.Event()
+    attempted = threading.Event()
+    finished = threading.Event()
+    errors = []
+
+    class BlockingOutput(Vt100_Output):
+        render_thread = None
+        in_render_flush = False
+        overlap = False
+
+        def flush(self):
+            if threading.get_ident() == self.render_thread:
+                self.in_render_flush = True
+                entered.set()
+                assert release.wait(timeout=5)
+                try:
+                    super().flush()
+                finally:
+                    self.in_render_flush = False
+            else:
+                self.overlap |= self.in_render_flush
+                super().flush()
+
+    output = BlockingOutput(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+        app = make_app(monkeypatch, pipe_input, output)
+        app.main_session.output.write_raw("Rendered")
+
+        def render():
+            output.render_thread = threading.get_ident()
+            app.main_session.output.flush()
+
+        def change_title():
+            attempted.set()
+            try:
+                app.set_window_title("Title")
+            except BaseException as error:  # noqa: BLE001
+                errors.append(error)
+            finally:
+                finished.set()
+
+        renderer = threading.Thread(target=render, daemon=True)
+        worker = threading.Thread(target=contextvars.copy_context().run, args=(change_title,), daemon=True)
+        renderer.start()
+        try:
+            assert entered.wait(timeout=5)
+            worker.start()
+            assert attempted.wait(timeout=5)
+            finished.wait(timeout=0.1)
+        finally:
+            release.set()
+            renderer.join(timeout=5)
+            worker.join(timeout=5)
+            assert not renderer.is_alive()
+            assert not worker.is_alive()
+
+    assert not errors
+    assert not output.overlap
+    assert stream.getvalue() == "Rendered\x1b]2;Title\x07"
+
+
+def test_title_without_terminal_output_is_a_noop():
+    app = cmd2.Cmd(stdin=io.StringIO(), stdout=io.StringIO(), allow_cli_args=False)
+    app.set_window_title("No terminal")
+    assert app.stdout.getvalue() == ""
+    assert isinstance(app.main_session.output, DummyOutput)
+
+
+@pytest.mark.skipif(sys.platform != "win32", reason="Uses the native Win32 title backend")
+def test_title_uses_the_native_win32_output_backend(monkeypatch):
+    from prompt_toolkit.output.win32 import Win32Output, windll
+
+    # Exercise native set_title/flush without needing a physical console in CI.
+    stream = TtyStream()
+    output = Win32Output.__new__(Win32Output)
+    output.stdout = stream
+    output._buffer = []
+    titles = []
+    monkeypatch.setattr(windll.kernel32, "SetConsoleTitleW", lambda title: titles.append(title) or 1)
+    app = cmd2.Cmd(stdin=io.StringIO(), stdout=io.StringIO(), allow_cli_args=False)
+    app.main_session.app.output = output
+    app.main_session.app.renderer.output = output
+    app.main_session = app.main_session
+    app.set_window_title("Working…")
+    assert titles == ["Working…"]
+
+
+def test_replacing_main_session_keeps_rendering_and_titles_on_the_new_output():
+    stream = TtyStream()
+    foreign = TtyStream()
+    observed = []
+    errors = []
+    workers = []
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = cmd2.Cmd(stdin=io.StringIO(), stdout=io.StringIO(), allow_cli_args=False)
+        app.main_session = PromptSession(input=pipe_input, output=title_output(stream))
+        assert app.main_session.output is app.main_session.app.renderer.output
+        app.set_window_title("Before")
+        assert "\x1b]2;Before\x07" in stream.getvalue()
+        sent = False
+
+        def update_title():
+            try:
+                app.set_window_title("Replacement prompt")
+                observed.append("\x1b]2;Replacement prompt\x07" in stream.getvalue())
+            except BaseException as error:  # noqa: BLE001
+                errors.append(error)
+            finally:
+                app.main_session.app.loop.call_soon_threadsafe(lambda: app.main_session.app.exit(result="done"))
+
+        def after_render(_sender):
+            nonlocal sent
+            if not sent:
+                sent = True
+                worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True)
+                workers.append(worker)
+                worker.start()
+
+        app.main_session.app.after_render += after_render
+        assert app._read_raw_input("replacement> ", app.main_session, set_exception_handler=False) == "done"
+        for worker in workers:
+            worker.join(timeout=5)
+            assert not worker.is_alive()
+        app.set_window_title("After")
+
+    assert not errors
+    assert observed == [True]
+    assert "\x1b]2;After\x07" in stream.getvalue()
+    assert foreign.getvalue() == ""