diff --git a/CHANGELOG.md b/CHANGELOG.md
index a463c146d..2b3e127c5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -28,6 +28,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.
 
 ## 4.3.1 (October 24, 2026)
 
diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py
index de2e140a4..773b6d6bb 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
+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
@@ -84,7 +85,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,
@@ -220,6 +221,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."""
 
@@ -4792,10 +4798,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:
@@ -5884,12 +5900,42 @@ 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, including any active select() menu.
+        Worker threads serving a custom app session must inherit that session's context.
+
         :param title: the new window title
         """
-        set_title(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.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/features/prompt.md b/docs/features/prompt.md
index 479ef9e55..c1dd1a2d9 100644
--- a/docs/features/prompt.md
+++ b/docs/features/prompt.md
@@ -54,6 +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`.
+
 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..faf0c240b 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, get_app_session
+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,339 @@ 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