diff --git a/CHANGELOG.md b/CHANGELOG.md
index b0ac68c3d..2bc9b730b 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
@@ -32,6 +34,12 @@
       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. The
+      title is now flushed before the method returns when no prompt is running. While a prompt or
+      `select()` menu is running, it is written on that prompt's event loop so it doesn't interleave
+      with rendering, and output errors are ignored so they can't disrupt the prompt. Titles are
+      still written to a terminal stdout when prompts use a dummy output, such as when stdin isn't a
+      terminal.
 
 ## 4.3.1 (October 24, 2026)
 
diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index de2e140a4..28ebf48df 100644
--- a/cmd2/cmd2.py
+++ b/cmd2/cmd2.py
@@ -29,6 +29,7 @@
 # import this module, many of these imports are lazy-loaded
 # i.e. we only import the module when we use it.
 import argparse
+import asyncio
 import contextlib
 import copy
 import dataclasses
@@ -71,7 +72,7 @@
     filters,
     print_formatted_text,
 )
-from prompt_toolkit.application import create_app_session, get_app
+from prompt_toolkit.application import Application, AppSession, 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
@@ -82,9 +83,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 +221,11 @@ def __init__(self, msg: str = "") -> None:
     from prompt_toolkit.buffer import Buffer
 
 
+# Errors raised when prompt-toolkit can't use a terminal. For example, GitHub Actions on Windows reports
+# isatty() as True but has no real console.
+_TERMINAL_INIT_ERRORS = (NoConsoleScreenBufferError, AttributeError, ValueError)
+
+
 class _SavedCmd2Env:
     """cmd2 environment settings that are backed up when entering an interactive Python shell."""
 
@@ -689,6 +695,26 @@ def __init__(
         self._alert_thread: threading.Thread | None = None
         self._alert_prompt_timestamp: float = 0.0  # Uses time.monotonic()
 
+        # Members related to setting the window title. The lock is reentrant so a signal handler which sets the title
+        # while the main thread is writing one doesn't deadlock. Such a nested call only records its title, and the
+        # outer call writes it next.
+        self._window_title_lock = threading.RLock()
+        self._pending_window_title: str | None = None
+        self._writing_window_title = False
+
+        # The prompt loop on which a title callback is queued but hasn't run yet. Later titles set while it's queued
+        # only replace the pending title, which that callback writes.
+        self._window_title_callback_loop: asyncio.AbstractEventLoop | None = None
+
+        # The stdout this app was created with. When the main session uses a DummyOutput, titles are written to it if
+        # it's a terminal, including while a command's output is redirected. See _get_window_title_fallback_output().
+        self._window_title_stdout: TextIO = self.stdout
+        self._window_title_fallback_checked = False
+        self._window_title_fallback_output: Output | None = None
+
+        # The AppSession of a select() menu while it is displayed, so window titles can find its event loop.
+        self._select_app_session: AppSession | None = None
+
         # Add functions decorated to be subcommands
         self._register_subcommands(self)
 
@@ -838,7 +864,7 @@ def _create_main_session(
                     kwargs["output"] = create_output(stdout=self.stdout)
                 return PromptSession(**kwargs)
 
-            except (NoConsoleScreenBufferError, AttributeError, ValueError):
+            except _TERMINAL_INIT_ERRORS:
                 # Fallback to dummy input/output if PromptSession initialization fails.
                 # This can happen in some CI environments (like GitHub Actions on Windows)
                 # where isatty() is True but there is no real console.
@@ -3659,14 +3685,19 @@ def _read_raw_input(
             if not callable(prompt):
                 prompt = pt_filter_style(prompt)
 
-            with patch_stdout():
-                try:
-                    # Set this session as the active one for UI/completion logic.
-                    self.active_session = session
-                    return session.prompt(prompt, **prompt_kwargs)
-                finally:
-                    # Revert back to the main session.
-                    self.active_session = self.main_session
+            try:
+                with patch_stdout():
+                    try:
+                        # Set this session as the active one for UI/completion logic.
+                        self.active_session = session
+                        return session.prompt(prompt, **prompt_kwargs)
+                    finally:
+                        # Revert back to the main session.
+                        self.active_session = self.main_session
+            finally:
+                # Write any title whose callback the prompt's event loop didn't run. Do this after patch_stdout()
+                # exits so the write can't race with its writer thread on the same output.
+                self._flush_pending_window_title_quietly()
 
         # We're not at a terminal, so we're likely reading from a file or a pipe.
         prompt_obj = prompt if isinstance(prompt, (ANSI, str)) else prompt()
@@ -4794,8 +4825,13 @@ def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], p
         if self._is_tty_session(self.main_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 app_session:
+                        self._select_app_session = app_session
+                        try:
+                            result = choice(message=prompt, options=fulloptions)
+                        finally:
+                            self._select_app_session = None
+                            self._flush_pending_window_title_quietly()
                     if result is not None:
                         return result
             except KeyboardInterrupt:
@@ -5883,13 +5919,125 @@ 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 the title of this application's terminal.
+
+        The title is written to the output used by the main prompt, custom input prompts, and
+        select() menus. That is a dummy output when stdin or stdout isn't a terminal, or when the
+        terminal couldn't be set up. In that case, the title is written to the stdout this app was
+        created with instead, if that is a terminal. Otherwise, no title is written.
+
+        This can be called from any thread. If no prompt is running, then the title is written and
+        flushed before this returns, and output errors propagate to the caller. Such a direct write
+        from another thread isn't synchronized with other writers to the terminal, such as a prompt
+        which is just starting or finishing.
+
+        While a prompt is running, the title is written on the prompt's event loop, even when this is
+        called on that loop, so it doesn't interleave with rendering. It appears once the loop runs its
+        next callback, which can be delayed while the loop is busy, such as in a slow completer. If
+        several titles are set before then, only the latest is written. Output errors are ignored so
+        they can't disrupt the prompt.
 
         :param title: the new window title
         """
-        set_title(title)
+        with self._window_title_lock:
+            self._pending_window_title = title
+            loop = self._get_prompt_loop()
+            if loop is not None:
+                if self._window_title_callback_loop is loop:
+                    # A callback queued on this loop hasn't run yet, and it will write this title.
+                    return
+
+                try:
+                    loop.call_soon_threadsafe(self._flush_pending_window_title_quietly)
+                except RuntimeError:
+                    # The loop closed after we found it, so the prompt is no longer rendering.
+                    pass
+                else:
+                    # If the prompt finishes before the loop runs the callback, then the prompt's
+                    # cleanup writes the pending title.
+                    self._window_title_callback_loop = loop
+                    return
+
+            # Keep ownership until the direct write finishes, so another caller or prompt cleanup cannot consume
+            # this request and receive or suppress its output error.
+            self._flush_pending_window_title()
+
+    def _get_prompt_loop(self) -> asyncio.AbstractEventLoop | None:
+        """Return the event loop of the prompt-toolkit application drawing on this terminal, if one is running.
+
+        An application's loop is set for its whole run, including startup and teardown, when its is_running flag
+        is already cleared but the loop can still write to the terminal.
+        """
+        apps: list[Application[Any]] = [self.active_session.app]
+
+        # The main thread clears these when select() finishes, so read each one only once.
+        select_app_session = self._select_app_session
+        if select_app_session is not None:
+            select_app = select_app_session.app
+            if select_app is not None:
+                apps.append(select_app)
+
+        for app in apps:
+            loop = app.loop
+            if loop is not None:
+                return loop
+        return None
+
+    def _get_window_title_fallback_output(self) -> Output | None:
+        """Return an output for the stdout this app was created with, or None if that isn't a terminal.
+
+        Window titles use it when the main session has a DummyOutput. It's created the first time a title needs it.
+        This isn't a cached_property because inspect.getmembers() evaluates properties while this app is constructed.
+        """
+        if not self._window_title_fallback_checked:
+            self._window_title_fallback_checked = True
+
+            # A stream without isatty(), or a closed one, raises AttributeError or ValueError here.
+            with contextlib.suppress(*_TERMINAL_INIT_ERRORS):
+                if self._window_title_stdout.isatty():
+                    self._window_title_fallback_output = create_output(stdout=self._window_title_stdout)
+
+        return self._window_title_fallback_output
+
+    def _get_window_title_output(self) -> Output:
+        """Return the output which window titles are written to."""
+        output = self.main_session.output
+        if isinstance(output, DummyOutput) and (fallback := self._get_window_title_fallback_output()) is not None:
+            return fallback
+        return output
+
+    def _flush_pending_window_title(self) -> None:
+        """Write and flush the most recent title passed to set_window_title(), if it hasn't been written yet."""
+        with self._window_title_lock:
+            if self._writing_window_title:
+                # This thread is already writing a title, such as when a signal handler calls set_window_title().
+                # Writing now would nest inside that write, so leave the title pending for the outer call.
+                return
+
+            self._writing_window_title = True
+            try:
+                # Keep going until no newer title arrived during the last write, so the latest title always wins.
+                while (title := self._pending_window_title) is not None:
+                    self._pending_window_title = None
+                    output = self._get_window_title_output()
+                    output.set_title(title)
+                    output.flush()
+            finally:
+                self._writing_window_title = False
+
+    def _flush_pending_window_title_quietly(self) -> None:
+        """Flush a pending window title on a prompt's event loop or during its cleanup, ignoring output errors.
+
+        An error here would otherwise disrupt the prompt or replace its result. Writing to a closed stream raises
+        ValueError rather than OSError.
+        """
+        with self._window_title_lock:
+            # Either this is the queued callback, or the prompt finished and its loop won't run that callback.
+            # Either way, the next title set during a prompt needs a new callback.
+            self._window_title_callback_loop = None
+            with contextlib.suppress(OSError, ValueError):
+                self._flush_pending_window_title()
 
     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 479ef9e55..59aeeed05 100644
--- a/docs/features/prompt.md
+++ b/docs/features/prompt.md
@@ -54,6 +54,26 @@ 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.
+
+The method can be called from any thread. If no prompt is running, then the title is flushed before
+the method returns, and output errors such as a closed terminal propagate to the caller. Such a
+direct write from another thread isn't synchronized with other writers to the terminal, such as a
+prompt which is just starting or finishing.
+
+While a prompt or `select()` menu is running, the title is written on that prompt's event loop, even
+when the method is called on that loop, so it doesn't interleave with rendering. The title appears
+once the loop runs its next callback, which can be delayed while the loop is busy, for example in a
+slow completer. If several titles are set before then, only the latest is written. Output errors are
+ignored in that case so they can't disrupt the prompt.
+
+Prompts use a dummy output when stdin or stdout isn't a terminal, or when the terminal couldn't be
+set up. In that case, titles are written to the stdout the application was created with instead, if
+that is a terminal, including while a command's output is redirected. Otherwise, no title is
+emitted.
+
 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/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_window_title.py b/tests/test_window_title.py
new file mode 100644
index 000000000..1b0d598fc
--- /dev/null
+++ b/tests/test_window_title.py
@@ -0,0 +1,828 @@
+"""Window title behavior through real prompt-toolkit output and applications."""
+
+import asyncio
+import contextlib
+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.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 PromptSession
+
+import cmd2
+
+
+class TtyStream(io.StringIO):
+    def isatty(self):
+        return True
+
+
+class TitleOutput(Vt100_Output):
+    """VT100 output which records the event loop, if any, that wrote each title."""
+
+    def __init__(self, stream):
+        super().__init__(stream, lambda: Size(rows=24, columns=80), enable_cpr=False)
+        self.title_loops = {}
+
+    def set_title(self, title):
+        try:
+            self.title_loops[title] = asyncio.get_running_loop()
+        except RuntimeError:
+            self.title_loops[title] = None
+        super().set_title(title)
+
+
+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 TitleOutput(stream)
+
+
+def start_worker(target, *args):
+    """Run target on a new thread with an empty context.
+
+    The returned future's result() re-raises any exception from target, so tests don't need a blind except.
+    """
+    executor = ThreadPoolExecutor(max_workers=1)
+    try:
+        return executor.submit(contextvars.Context().run, target, *args)
+    finally:
+        executor.shutdown(wait=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() == ""
+
+
+def test_title_uses_the_ambient_app_session_output_for_sys_stdout(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    monkeypatch.setattr(sys, "stdout", TtyStream())
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+        monkeypatch.setattr("cmd2.cmd2.create_input", lambda **_kwargs: pipe_input)
+        app = cmd2.Cmd(stdin=TtyStream(), allow_cli_args=False)
+        assert app.main_session.output is output
+        app.set_window_title("Ambient")
+
+    assert stream.getvalue().endswith("\x1b]2;Ambient\x07")
+
+
+@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()
+    output = title_output(stream)
+    observed = []
+    workers = []
+    prompt_loops = []
+    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, output)
+        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())
+            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
+            prompt_loops.append(prompt_app.loop)
+            if from_worker:
+                workers.append(start_worker(update_title, prompt_app))
+            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.result(timeout=5)
+        app.set_window_title("After")
+
+    assert result == "done"
+    # While the prompt runs, the title is written by a callback on its event loop. A worker's title may already be in the
+    # stream when set_window_title() returns. A call on the loop itself returns before the loop can run the callback.
+    assert len(observed) == 1
+    if not from_worker:
+        assert observed == [False]
+    assert output.title_loops["During"] is prompt_loops[0]
+    assert output.title_loops["Before"] is None
+    assert output.title_loops["After"] is None
+    assert "\x1b]2;Before\x07" in stream.getvalue()
+    assert "\x1b]2;During\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()
+    output = title_output(stream)
+    menu_loops = []
+    run = Application.run
+
+    def run_menu(menu, *args, **kwargs):
+        sent = False
+        workers = []
+
+        def update_title():
+            try:
+                app.set_window_title("Menu")
+            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
+                menu_loops.append(menu.loop)
+                workers.append(start_worker(update_title))
+
+        menu.after_render += after_render
+        try:
+            return run(menu, *args, set_exception_handler=False, **kwargs)
+        finally:
+            for worker in workers:
+                worker.result(timeout=5)
+
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)):
+        app = make_app(monkeypatch, pipe_input, 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 app._select_app_session is None
+        app.set_window_title("Finished")
+
+    assert output.title_loops["Menu"] is menu_loops[0]
+    assert output.title_loops["Finished"] is None
+    assert "\x1b]2;Menu\x07" in stream.getvalue()
+    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"])
+        assert app._select_app_session is None
+        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() == ""
+
+
+class FailingOutput(TitleOutput):
+    """Output which fails in either set_title() or flush() when the title is 'Fail'."""
+
+    def __init__(self, stream, operation):
+        super().__init__(stream)
+        self.operation = operation
+        self.fail_flush = False
+
+    def set_title(self, title):
+        if self.operation == "set_title" and title == "Fail":
+            raise OSError("terminal closed")
+        super().set_title(title)
+        if title == "Fail":
+            self.fail_flush = self.operation == "flush"
+
+    def flush(self):
+        if self.fail_flush:
+            self.fail_flush = False
+            raise OSError("terminal closed")
+        super().flush()
+
+
+@pytest.mark.parametrize("operation", ["set_title", "flush"])
+def test_title_output_errors_reach_the_caller_without_a_prompt(monkeypatch, operation):
+    output = FailingOutput(TtyStream(), operation)
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, output)
+        with pytest.raises(OSError, match="terminal closed"):
+            app.set_window_title("Fail")
+
+        # A failed title isn't retried.
+        app._flush_pending_window_title()
+        assert app._pending_window_title is None
+
+
+@pytest.mark.parametrize("closed_loop", [False, True])
+@pytest.mark.parametrize("error_type", [OSError, ValueError])
+@pytest.mark.parametrize("operation", ["set_title", "flush"])
+def test_prompt_cleanup_cannot_swallow_a_direct_title_error(monkeypatch, closed_loop, error_type, operation):
+    output = title_output(TtyStream())
+    flush_entered = threading.Event()
+    release_flush = threading.Event()
+    caller_threads = []
+    failed_threads = []
+
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, output)
+        if closed_loop:
+            loop = asyncio.new_event_loop()
+            loop.close()
+            monkeypatch.setattr(app.main_session.app, "loop", loop)
+        flush = app._flush_pending_window_title
+
+        def pause_direct_flush():
+            if threading.get_ident() == caller_threads[0]:
+                flush_entered.set()
+                assert release_flush.wait(timeout=5)
+            flush()
+
+        def fail_output(*_args):
+            failed_threads.append(threading.get_ident())
+            raise error_type("terminal closed")
+
+        def update_title():
+            caller_threads.append(threading.get_ident())
+            app.set_window_title("Fail")
+
+        monkeypatch.setattr(app, "_flush_pending_window_title", pause_direct_flush)
+        monkeypatch.setattr(output, operation, fail_output)
+        caller = start_worker(update_title)
+        cleanup = None
+        try:
+            assert flush_entered.wait(timeout=5)
+            cleanup = start_worker(app._flush_pending_window_title_quietly)
+            # If the caller released ownership before flushing, let cleanup consume the title first.
+            # Otherwise cleanup must wait for the caller; release it without waiting for a timeout.
+            if app._window_title_lock.acquire(blocking=False):
+                app._window_title_lock.release()
+                cleanup.result(timeout=5)
+        finally:
+            release_flush.set()
+
+        try:
+            with pytest.raises(error_type, match="terminal closed"):
+                caller.result(timeout=5)
+        finally:
+            if cleanup is not None:
+                cleanup.result(timeout=5)
+        app._flush_pending_window_title_quietly()
+
+    assert failed_threads == caller_threads
+    assert app._pending_window_title is None
+    assert not app._writing_window_title
+
+
+@pytest.mark.parametrize("operation", ["set_title", "flush"])
+@pytest.mark.parametrize("from_worker", [False, True])
+def test_title_output_errors_during_a_prompt(monkeypatch, operation, from_worker):
+    caught = []
+    loop_errors = []
+    workers = []
+    output = FailingOutput(TtyStream(), operation)
+    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:
+                workers.append(start_worker(update_title))
+            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.result(timeout=5)
+
+    # The title was written by an event loop callback, even when set on the loop itself. The callback ignores output
+    # errors, so neither the caller nor the loop's exception handler sees them, and they can't disrupt the prompt.
+    assert not caught
+    assert not loop_errors
+    assert app._pending_window_title is None
+
+
+@pytest.mark.parametrize("failing", [False, True])
+def test_title_is_written_when_the_prompt_finishes_before_running_the_callback(monkeypatch, failing):
+    stream = TtyStream()
+    output = FailingOutput(stream, "flush") if failing else title_output(stream)
+    title = "Fail" if failing else "Finished"
+    dropped = []
+    workers = []
+    with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+        app = make_app(monkeypatch, pipe_input, output)
+        prompt_app = app.main_session.app
+        sent = False
+
+        def update_title():
+            loop = prompt_app.loop
+            schedule = loop.call_soon_threadsafe
+
+            # Simulate the event loop stopping before it runs the title callback.
+            def drop_title_callback(callback, *args, **kwargs):
+                if getattr(callback, "__func__", None) is cmd2.Cmd._flush_pending_window_title_quietly:
+                    dropped.append(callback)
+                    return None
+                return schedule(callback, *args, **kwargs)
+
+            monkeypatch.setattr(loop, "call_soon_threadsafe", drop_title_callback)
+            app.set_window_title(title)
+            schedule(lambda: prompt_app.exit(result="done"))
+
+        def after_render(_sender):
+            nonlocal sent
+            if not sent:
+                sent = True
+                workers.append(start_worker(update_title))
+
+        prompt_app.after_render += after_render
+        assert app._read_raw_input("prompt> ", app.main_session, set_exception_handler=False) == "done"
+        for worker in workers:
+            worker.result(timeout=5)
+
+    assert len(dropped) == 1
+    assert app._pending_window_title is None
+    assert output.title_loops[title] is None
+    if not failing:
+        # The prompt's cleanup wrote the title after the prompt finished rendering.
+        assert stream.getvalue().endswith("\x1b]2;Finished\x07")
+    # Otherwise the cleanup's output error didn't replace the prompt's result, which the assertion on "done" checked.
+
+
+def test_title_error_during_select_cleanup_keeps_the_interrupt(monkeypatch):
+    output = FailingOutput(TtyStream(), "flush")
+
+    def interrupt_with_pending_title(**_kwargs):
+        # Leave a title for the cleanup, as if the menu's loop stopped before running the title callback.
+        app._pending_window_title = "Fail"
+        raise KeyboardInterrupt
+
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, output)
+        monkeypatch.setattr("cmd2.cmd2.choice", interrupt_with_pending_title)
+        with pytest.raises(KeyboardInterrupt):
+            app.select(["choice"])
+
+    assert app._pending_window_title is None
+    assert app._select_app_session is None
+
+
+def test_prompt_loop_ignores_a_select_session_without_an_app(monkeypatch):
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, title_output(TtyStream()))
+        with create_app_session(input=DummyInput(), output=DummyOutput()) as app_session:
+            app._select_app_session = app_session
+            assert app_session.app is None
+            assert app._get_prompt_loop() is None
+
+
+def test_title_set_while_writing_a_title_is_written_next(monkeypatch):
+    stream = TtyStream()
+
+    class ReentrantOutput(TitleOutput):
+        flushing = False
+        nested_flush = False
+
+        def set_title(self, title):
+            if title == "Outer":
+                # Like a signal handler which sets the title while this thread is writing one.
+                app.set_window_title("Inner")
+            super().set_title(title)
+
+        def flush(self):
+            self.nested_flush |= self.flushing
+            self.flushing = True
+            try:
+                if "\x1b]2;Outer\x07" in "".join(self._buffer):
+                    # Like a signal handler which runs while this thread is writing to the stream.
+                    app.set_window_title("During flush")
+                super().flush()
+            finally:
+                self.flushing = False
+
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, ReentrantOutput(stream))
+
+        # Run in a thread so a deadlock fails the test instead of hanging it.
+        start_worker(app.set_window_title, "Outer").result(timeout=5)
+
+    # The nested calls didn't write inside the outer write. "During flush" replaced "Inner" before it was written, so
+    # only the latest title was written after "Outer".
+    assert not app.main_session.output.nested_flush
+    assert stream.getvalue() == "\x1b]2;Outer\x07\x1b]2;During flush\x07"
+    assert app._pending_window_title is None
+    assert not app._writing_window_title
+
+
+def test_quiet_title_flush_ignores_a_closed_stream(monkeypatch):
+    stream = TtyStream()
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+        stream.close()
+        app._pending_window_title = "Closed"
+        app._flush_pending_window_title_quietly()
+
+    assert app._pending_window_title is None
+    assert not app._writing_window_title
+
+
+def test_prompt_cleanup_writes_the_title_after_patch_stdout_exits(monkeypatch):
+    events = []
+
+    class RecordingOutput(TitleOutput):
+        def set_title(self, title):
+            events.append(f"title {title}")
+            super().set_title(title)
+
+    @contextlib.contextmanager
+    def recording_patch_stdout():
+        events.append("patch_stdout enter")
+        try:
+            yield
+        finally:
+            events.append("patch_stdout exit")
+
+    def prompt_with_pending_title(*_args, **_kwargs):
+        # Leave a title for the cleanup, as if the prompt's loop stopped before running the title callback.
+        app._pending_window_title = "Cleanup"
+        return "done"
+
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, RecordingOutput(TtyStream()))
+        monkeypatch.setattr("cmd2.cmd2.patch_stdout", recording_patch_stdout)
+        monkeypatch.setattr(app.main_session, "prompt", prompt_with_pending_title)
+        assert app._read_raw_input("prompt> ", app.main_session) == "done"
+
+    assert events == ["patch_stdout enter", "patch_stdout exit", "title Cleanup"]
+
+
+def test_title_falls_back_to_a_terminal_stdout_when_stdin_is_not_a_terminal(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    created = []
+
+    def create_title_output(**kwargs):
+        created.append(kwargs["stdout"])
+        return output
+
+    monkeypatch.setattr("cmd2.cmd2.create_output", create_title_output)
+    app = cmd2.Cmd(stdin=io.StringIO(), stdout=stream, allow_cli_args=False)
+    assert isinstance(app.main_session.output, DummyOutput)
+
+    # The fallback output is only created once a title needs it.
+    assert not created
+    app.set_window_title("Piped input")
+    app.set_window_title("Again")
+    assert created == [stream]
+    assert stream.getvalue() == "\x1b]2;Piped input\x07\x1b]2;Again\x07"
+
+    # Titles still reach the terminal while a command's output is redirected.
+    app.stdout = io.StringIO()
+    app.set_window_title("Redirected")
+    assert stream.getvalue().endswith("\x1b]2;Redirected\x07")
+    assert app.stdout.getvalue() == ""
+
+    # A replacement session with a real output takes precedence over the fallback.
+    replacement = TtyStream()
+    with create_pipe_input() as pipe_input:
+        app.main_session = PromptSession(input=pipe_input, output=title_output(replacement))
+        app.set_window_title("Replacement")
+    assert replacement.getvalue().endswith("\x1b]2;Replacement\x07")
+    assert "Replacement" not in stream.getvalue()
+
+
+def test_title_falls_back_when_a_replacement_session_has_no_terminal_output(monkeypatch):
+    stream = TtyStream()
+    fallback = TtyStream()
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, title_output(stream))
+    # The fallback output hasn't been created, since no title has needed it.
+    assert not app._window_title_fallback_checked
+
+    app.main_session = PromptSession(input=DummyInput(), output=DummyOutput())
+    monkeypatch.setattr("cmd2.cmd2.create_output", lambda **_kwargs: title_output(fallback))
+    app.set_window_title("Fallback")
+    assert fallback.getvalue() == "\x1b]2;Fallback\x07"
+    assert stream.getvalue() == ""
+
+
+def test_title_fallback_is_skipped_when_the_terminal_output_cannot_be_created(monkeypatch):
+    attempts = []
+
+    def no_console(**_kwargs):
+        attempts.append(True)
+        raise ValueError("no console")
+
+    monkeypatch.setattr("cmd2.cmd2.create_output", no_console)
+    app = cmd2.Cmd(stdin=io.StringIO(), stdout=TtyStream(), allow_cli_args=False)
+    app.set_window_title("Nowhere")
+    app.set_window_title("Still nowhere")
+    assert app._window_title_fallback_output is None
+    assert app.stdout.getvalue() == ""
+
+    # A failure isn't retried for every title.
+    assert len(attempts) == 1
+
+
+def test_title_is_written_directly_when_the_prompt_loop_is_closed(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    loop = asyncio.new_event_loop()
+    loop.close()
+    with create_pipe_input() as pipe_input:
+        app = make_app(monkeypatch, pipe_input, output)
+        monkeypatch.setattr(app.main_session.app, "loop", loop)
+        app.set_window_title("Closed loop")
+
+    assert stream.getvalue() == "\x1b]2;Closed loop\x07"
+    assert app._pending_window_title is None
+
+
+def test_title_is_scheduled_while_a_finishing_prompt_still_has_its_loop(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    loop = asyncio.new_event_loop()
+    try:
+        with create_pipe_input() as pipe_input:
+            app = make_app(monkeypatch, pipe_input, output)
+            prompt_app = app.main_session.app
+            # Like prompt-toolkit's teardown, which clears is_running before its loop stops writing to the terminal.
+            assert not prompt_app.is_running
+            monkeypatch.setattr(prompt_app, "loop", loop)
+            app.set_window_title("Teardown")
+            assert stream.getvalue() == ""
+
+            loop.call_soon(loop.stop)
+            loop.run_forever()
+    finally:
+        loop.close()
+
+    assert stream.getvalue() == "\x1b]2;Teardown\x07"
+    assert output.title_loops["Teardown"] is loop
+
+
+def test_titles_set_before_the_loop_runs_share_one_callback(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    loop = asyncio.new_event_loop()
+    scheduled = []
+    schedule = loop.call_soon_threadsafe
+
+    def count_callbacks(callback, *args, **kwargs):
+        scheduled.append(callback)
+        return schedule(callback, *args, **kwargs)
+
+    try:
+        with create_pipe_input() as pipe_input:
+            app = make_app(monkeypatch, pipe_input, output)
+            monkeypatch.setattr(app.main_session.app, "loop", loop)
+            monkeypatch.setattr(loop, "call_soon_threadsafe", count_callbacks)
+            for title in ("One", "Two", "Three"):
+                app.set_window_title(title)
+            assert len(scheduled) == 1
+
+            loop.call_soon(loop.stop)
+            loop.run_forever()
+
+            # Only the latest title was written, and the next title gets a new callback.
+            assert stream.getvalue() == "\x1b]2;Three\x07"
+            assert app._window_title_callback_loop is None
+            app.set_window_title("Four")
+            assert len(scheduled) == 2
+
+            loop.call_soon(loop.stop)
+            loop.run_forever()
+    finally:
+        loop.close()
+
+    assert stream.getvalue() == "\x1b]2;Three\x07\x1b]2;Four\x07"
+
+
+def test_prompt_cleanup_clears_a_callback_its_loop_never_ran(monkeypatch):
+    stream = TtyStream()
+    output = title_output(stream)
+    stopped_loop = asyncio.new_event_loop()
+    try:
+        # Give patch_stdout() a VT100 output, since a Windows CI runner has no console for the default one.
+        with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=output):
+            app = make_app(monkeypatch, pipe_input, output)
+            prompt_app = app.main_session.app
+
+            def prompt_with_unrun_callback(*_args, **_kwargs):
+                # The title callback is queued on a loop which stops without running it.
+                monkeypatch.setattr(prompt_app, "loop", stopped_loop)
+                app.set_window_title("Queued")
+                monkeypatch.setattr(prompt_app, "loop", None)
+                return "done"
+
+            monkeypatch.setattr(app.main_session, "prompt", prompt_with_unrun_callback)
+            assert app._read_raw_input("prompt> ", app.main_session) == "done"
+    finally:
+        stopped_loop.close()
+
+    # The prompt's cleanup wrote the title and cleared the queued callback's loop, so later prompts schedule their own.
+    assert stream.getvalue() == "\x1b]2;Queued\x07"
+    assert app._window_title_callback_loop is None
+
+
+class NoIsattyStream:
+    """A minimal file-like stdout without isatty()."""
+
+    def __init__(self):
+        self.written = []
+
+    def write(self, text):
+        self.written.append(text)
+        return len(text)
+
+    def flush(self):
+        pass
+
+
+@pytest.mark.parametrize("closed", [False, True])
+def test_stdout_which_cannot_report_a_terminal_does_not_break_titles(closed):
+    if closed:
+        # isatty() on a closed stream raises ValueError.
+        stdout = io.StringIO()
+        app = cmd2.Cmd(stdin=io.StringIO(), stdout=stdout, allow_cli_args=False)
+        stdout.close()
+    else:
+        stdout = NoIsattyStream()
+        app = cmd2.Cmd(stdin=io.StringIO(), stdout=stdout, allow_cli_args=False)
+
+    app.set_window_title("Nowhere")
+    app.set_window_title("Still nowhere")
+    # The check ran once and found no terminal.
+    assert app._window_title_fallback_checked
+    assert app._window_title_fallback_output is None
+    if not closed:
+        assert stdout.written == []
+
+
+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.set_window_title("Working…")
+    assert titles == ["Working…"]
+
+
+def test_replacing_main_session_keeps_titles_on_the_new_output():
+    stream = TtyStream()
+    foreign = TtyStream()
+    output = title_output(stream)
+    workers = []
+    prompt_loops = []
+    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=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")
+            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
+                prompt_loops.append(app.main_session.app.loop)
+                workers.append(start_worker(update_title))
+
+        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.result(timeout=5)
+        app.set_window_title("After")
+
+    assert output.title_loops["Replacement prompt"] is prompt_loops[0]
+    assert "\x1b]2;Replacement prompt\x07" in stream.getvalue()
+    assert "\x1b]2;After\x07" in stream.getvalue()
+    assert foreign.getvalue() == ""