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 01/12] 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 02/12] 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 03/12] 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 04/12] 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() == "" From 70da13a8a6429cd69ed649a1c9c71a802f3ce5b2 Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:01:06 -0400 Subject: [PATCH 05/12] Write window titles on the running prompt's event loop instead of locking the output Replace the locked output proxy from #1783 with a smaller change confined to set_window_title(). The title is written and flushed on main_session.output. When another thread calls it while a prompt or select() menu is running, the write is scheduled on that prompt's event loop so it doesn't interleave with rendering. A pending title is written when the prompt finishes in case the loop stops before running the callback. This drops the _LockedOutput proxy, the main_session property setter, and the forced per-Cmd output driver, restoring prompt-toolkit's dumb-terminal fallback, the ambient AppSession output when stdout is sys.stdout, and isinstance checks on the renderer's output. --- CHANGELOG.md | 6 +- cmd2/cmd2.py | 128 +++++++++-------- docs/features/prompt.md | 9 +- tests/test_cmd2.py | 4 - tests/test_window_title.py | 275 ++++++++++++++++++++----------------- 5 files changed, 224 insertions(+), 198 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a0f57142..5b58d50a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,9 +34,9 @@ it directly would then have retained current-session entries hidden from unbounded ranges and searches. Interactive `history` use was unaffected, since `cmd2` only truncates when persisting history at exit. - - Fixed `set_window_title()` leaving title changes buffered until the next prompt render. Title - writes and flushes are serialized with rendering on the application's output driver, including - during prompt setup and teardown. Output errors propagate to the calling thread. + - Fixed `set_window_title()` leaving title changes buffered until the next prompt render. The + title is now flushed right away. When another thread sets it while a prompt or `select()` menu + is running, it is written on that prompt's event loop so it doesn't interleave with rendering. ## 4.3.1 (October 24, 2026) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index a46d8cea2..b4766a592 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 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,7 +83,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, Output, create_output +from prompt_toolkit.output import DummyOutput, create_output from prompt_toolkit.patch_stdout import patch_stdout from prompt_toolkit.shortcuts import CompleteStyle, PromptSession, choice from prompt_toolkit.styles import DynamicStyle @@ -220,37 +221,6 @@ def __init__(self, msg: str = "") -> None: from prompt_toolkit.buffer import Buffer -class _LockedOutput: - """Serialize prompt rendering and synchronous title changes on one output driver.""" - - def __init__(self, output: Output) -> None: - """Wrap the output without changing its terminal or rendering behavior.""" - self._output = output - self._lock = threading.RLock() - - def __getattr__(self, name: str) -> Any: - """Forward output properties and serialize every output method.""" - attribute = getattr(self._output, name) - if not callable(attribute): - return attribute - - def locked_call(*args: Any, **kwargs: Any) -> Any: - with self._lock: - return attribute(*args, **kwargs) - - return locked_call - - def set_title_and_flush(self, title: str) -> None: - """Keep the title write and flush together, including during prompt setup and teardown.""" - with self._lock: - self._output.set_title(title) - self._output.flush() - - -# Like prompt-toolkit's Windows10_Output, this proxy implements Output by delegation. -Output.register(_LockedOutput) - - class _SavedCmd2Env: """cmd2 environment settings that are backed up when entering an interactive Python shell.""" @@ -720,6 +690,13 @@ 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 + self._window_title_lock = threading.Lock() + self._pending_window_title: str | 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) @@ -825,20 +802,6 @@ 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, *, @@ -879,8 +842,8 @@ def _create_main_session( try: if self.stdin != sys.stdin: kwargs["input"] = create_input(stdin=self.stdin) - # Each Cmd owns its driver, so another app session cannot share its render buffer. - kwargs["output"] = create_output(stdout=self.stdout) + if self.stdout != sys.stdout: + kwargs["output"] = create_output(stdout=self.stdout) return PromptSession(**kwargs) except (NoConsoleScreenBufferError, AttributeError, ValueError): @@ -3712,6 +3675,7 @@ def _read_raw_input( finally: # Revert back to the main session. self.active_session = self.main_session + self._flush_pending_window_title() # 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() @@ -4839,8 +4803,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() if result is not None: return result except KeyboardInterrupt: @@ -5929,20 +5898,57 @@ def add_alert( self._alert_condition.notify_all() def set_window_title(self, title: str) -> None: - """Set and immediately flush the title of this application's terminal. + """Set the title of this application's terminal and flush it right away. - 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. + The title is written to the same output as the main prompt, custom input prompts, and + select() menus. It is safe to call from any thread. While one of those prompts is running, + the title is written on the prompt's event loop so it doesn't interleave with rendering. :param title: the new window title """ - output = self.main_session.output - if isinstance(output, _LockedOutput): - output.set_title_and_flush(title) - else: - # Non-interactive sessions use a DummyOutput, which needs no synchronization. + with self._window_title_lock: + self._pending_window_title = title + loop = self._get_prompt_loop() + if loop is not None: + try: + on_loop = asyncio.get_running_loop() is loop + except RuntimeError: + on_loop = False + + if not on_loop: + try: + loop.call_soon_threadsafe(self._flush_pending_window_title) + 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. + return + + 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.""" + apps = [self.active_session.app] + if self._select_app_session is not None and self._select_app_session.app is not None: + apps.append(self._select_app_session.app) + + for app in apps: + loop = app.loop + if app.is_running and loop is not None: + return loop + return None + + 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: + title = self._pending_window_title + if title is None: + return + self._pending_window_title = None + + output = self.main_session.output output.set_title(title) output.flush() diff --git a/docs/features/prompt.md b/docs/features/prompt.md index 011517407..40bed13de 100644 --- a/docs/features/prompt.md +++ b/docs/features/prompt.md @@ -58,10 +58,11 @@ Call this method on your application instance, for example `app.set_window_title 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 method is safe to call from any thread, and the title is flushed right away. When another thread +calls it while a prompt or `select()` menu is running, the title is written on that prompt's event +loop so it doesn't interleave with rendering. In that case, output errors are reported by the +prompt's event loop. Otherwise, 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/tests/test_cmd2.py b/tests/test_cmd2.py index 487fbf7a8..2f39462d4 100644 --- a/tests/test_cmd2.py +++ b/tests/test_cmd2.py @@ -4842,8 +4842,6 @@ 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 ( @@ -4997,8 +4995,6 @@ 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 index ddd257fb8..d76e7f972 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -10,10 +10,9 @@ 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 import DummyOutput from prompt_toolkit.output.vt100 import Vt100_Output from prompt_toolkit.shortcuts import PromptSession @@ -25,6 +24,21 @@ 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: @@ -38,7 +52,7 @@ def make_app(monkeypatch, pipe_input, output): def title_output(stream): - return Vt100_Output(stream, lambda: Size(rows=24, columns=80), enable_cpr=False) + return TitleOutput(stream) def test_title_during_a_command_uses_the_instance_terminal(monkeypatch): @@ -50,9 +64,19 @@ def test_title_during_a_command_uses_the_instance_terminal(monkeypatch): 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) + + +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]) @@ -60,12 +84,14 @@ def test_title_during_a_command_uses_the_instance_terminal(monkeypatch): def test_title_before_during_and_after_a_real_prompt(monkeypatch, from_worker, custom_prompt): stream = TtyStream() foreign = TtyStream() + output = title_output(stream) observed = [] errors = [] 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, title_output(stream)) + app = make_app(monkeypatch, pipe_input, output) app.set_window_title("Before") sent = False @@ -83,6 +109,7 @@ def after_render(prompt_app): if sent: return sent = True + prompt_loops.append(prompt_app.loop) if from_worker: worker = threading.Thread(target=contextvars.Context().run, args=(update_title, prompt_app), daemon=True) workers.append(worker) @@ -107,8 +134,13 @@ def run_prompt(prompt_app, *args, **kwargs): assert result == "done" assert not errors - assert observed == [True] + # A worker's title is written asynchronously on the prompt's event loop, so only the loop thread sees it immediately. + assert observed == [not from_worker] + 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() == "" @@ -117,8 +149,9 @@ def run_prompt(prompt_app, *args, **kwargs): def test_worker_titles_during_a_real_select_and_after_cleanup(monkeypatch, interrupt): stream = TtyStream() foreign = TtyStream() - observed = [] + output = title_output(stream) errors = [] + menu_loops = [] run = Application.run def run_menu(menu, *args, **kwargs): @@ -128,7 +161,6 @@ def run_menu(menu, *args, **kwargs): 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: @@ -141,6 +173,7 @@ def after_render(_sender): nonlocal sent if not sent: sent = True + menu_loops.append(menu.loop) worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True) workers.append(worker) worker.start() @@ -154,17 +187,20 @@ def after_render(_sender): 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)) + 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 not errors - assert observed == [True] + 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() == "" @@ -226,6 +262,7 @@ def interrupt_before_application(**_kwargs): 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() @@ -233,31 +270,48 @@ def interrupt_before_application(**_kwargs): 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("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() +def test_title_output_errors_during_a_prompt(monkeypatch, operation, from_worker): 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) + 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 @@ -289,106 +343,76 @@ def after_render(_sender): worker.join(timeout=5) assert not worker.is_alive() - assert caught == ["terminal closed"] - assert not loop_errors + if from_worker: + # The title was written by an event loop callback, so the loop's exception handler reports the error. + assert not caught + assert len(loop_errors) == 1 + assert str(loop_errors[0]["exception"]) == "terminal closed" + else: + assert caught == ["terminal closed"] + assert not loop_errors -def test_title_is_not_queued_when_the_loop_stops_during_scheduling(monkeypatch): +def test_title_is_written_when_the_prompt_finishes_before_running_the_callback(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) - + dropped = [] + workers = [] 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 = [] + prompt_app = app.main_session.app + sent = False - 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() + def update_title(): + loop = prompt_app.loop + schedule = loop.call_soon_threadsafe - 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") + # 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: + dropped.append(callback) + return None + return schedule(callback, *args, **kwargs) - def render(): - output.render_thread = threading.get_ident() - app.main_session.output.flush() + monkeypatch.setattr(loop, "call_soon_threadsafe", drop_title_callback) + app.set_window_title("Finished") + schedule(lambda: prompt_app.exit(result="done")) - def change_title(): - attempted.set() - try: - app.set_window_title("Title") - except BaseException as error: # noqa: BLE001 - errors.append(error) - finally: - finished.set() + 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() - 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) + 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.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" + assert len(dropped) == 1 + assert app._pending_window_title is None + # The prompt's cleanup wrote the title after the prompt finished rendering. + assert stream.getvalue().endswith("\x1b]2;Finished\x07") + assert output.title_loops["Finished"] is None + + +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) + prompt_app = app.main_session.app + monkeypatch.setattr(prompt_app, "_is_running", True) + monkeypatch.setattr(prompt_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_without_terminal_output_is_a_noop(): @@ -411,22 +435,20 @@ def test_title_uses_the_native_win32_output_backend(monkeypatch): 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(): +def test_replacing_main_session_keeps_titles_on_the_new_output(): stream = TtyStream() foreign = TtyStream() - observed = [] + output = title_output(stream) errors = [] 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=title_output(stream)) - assert app.main_session.output is app.main_session.app.renderer.output + app.main_session = PromptSession(input=pipe_input, output=output) app.set_window_title("Before") assert "\x1b]2;Before\x07" in stream.getvalue() sent = False @@ -434,7 +456,6 @@ def test_replacing_main_session_keeps_rendering_and_titles_on_the_new_output(): 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: @@ -444,6 +465,7 @@ def after_render(_sender): nonlocal sent if not sent: sent = True + prompt_loops.append(app.main_session.app.loop) worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True) workers.append(worker) worker.start() @@ -456,6 +478,7 @@ def after_render(_sender): app.set_window_title("After") assert not errors - assert observed == [True] + 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() == "" From b0e8996b5ebdbf8a160f153806d354c350067c4c Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:11:12 -0400 Subject: [PATCH 06/12] Don't assert when a worker's title becomes visible during a prompt The loop thread can write the title before the worker checks the stream, which happens on free-threaded Python. Only a call on the loop thread is guaranteed to have written it before returning. --- tests/test_window_title.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/tests/test_window_title.py b/tests/test_window_title.py index d76e7f972..816aa4668 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -134,8 +134,11 @@ def run_prompt(prompt_app, *args, **kwargs): assert result == "done" assert not errors - # A worker's title is written asynchronously on the prompt's event loop, so only the loop thread sees it immediately. - assert observed == [not from_worker] + # A worker's title is written asynchronously on the prompt's event loop, so it may or may not be in the stream when + # set_window_title() returns. Only a call on the loop thread is guaranteed to have written it by then. + assert len(observed) == 1 + if not from_worker: + assert observed == [True] assert output.title_loops["During"] is prompt_loops[0] assert output.title_loops["Before"] is None assert output.title_loops["After"] is None From 5f46a34ee29c86884b1ab1f4deb8633de5c629aa Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:25:08 -0400 Subject: [PATCH 07/12] Harden window title writes found in review - Read the select() AppSession and its app once in _get_prompt_loop(), since the main thread clears them when select() finishes. - Ignore output errors when a title is written on a prompt's event loop or during prompt and select() cleanup, so they can't disrupt the prompt or replace its result or exception. - Use a reentrant lock so a signal handler which sets the title while one is being written doesn't deadlock. - Write titles to a terminal stdout when stdin isn't a terminal and the main session uses a DummyOutput. - Document that a title set from another thread appears once the prompt's loop runs its next callback. --- CHANGELOG.md | 6 +- cmd2/cmd2.py | 63 ++++++++++++++++----- docs/features/prompt.md | 16 ++++-- tests/test_window_title.py | 111 +++++++++++++++++++++++++++++++++---- 4 files changed, 164 insertions(+), 32 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b58d50a7..301c17b45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,8 +35,10 @@ 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 right away. When another thread sets it while a prompt or `select()` menu - is running, it is written on that prompt's event loop so it doesn't interleave with rendering. + title is now flushed before the method returns when no prompt is running. When another thread + sets it 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 when stdin isn't a terminal but stdout is. ## 4.3.1 (October 24, 2026) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index b4766a592..29ef6bce7 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -72,7 +72,7 @@ filters, print_formatted_text, ) -from prompt_toolkit.application import AppSession, 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 @@ -83,7 +83,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 @@ -690,10 +690,18 @@ 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 - self._window_title_lock = threading.Lock() + # 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. + self._window_title_lock = threading.RLock() self._pending_window_title: str | None = None + # When stdin isn't a terminal, the main session uses a DummyOutput. If stdout is a terminal, then + # window titles are still written to it. + self._window_title_fallback_output: Output | None = None + if isinstance(self.main_session.output, DummyOutput) and self.stdout.isatty(): + with contextlib.suppress(NoConsoleScreenBufferError, AttributeError, ValueError): + self._window_title_fallback_output = create_output(stdout=self.stdout) + # 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 @@ -3675,7 +3683,7 @@ def _read_raw_input( finally: # Revert back to the main session. self.active_session = self.main_session - self._flush_pending_window_title() + 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() @@ -4809,7 +4817,7 @@ def select(self, opts: str | Iterable[str] | Iterable[tuple[Any, str | None]], p result = choice(message=prompt, options=fulloptions) finally: self._select_app_session = None - self._flush_pending_window_title() + self._flush_pending_window_title_quietly() if result is not None: return result except KeyboardInterrupt: @@ -5898,11 +5906,18 @@ def add_alert( self._alert_condition.notify_all() def set_window_title(self, title: str) -> None: - """Set the title of this application's terminal and flush it right away. + """Set the title of this application's terminal. The title is written to the same output as the main prompt, custom input prompts, and - select() menus. It is safe to call from any thread. While one of those prompts is running, - the title is written on the prompt's event loop so it doesn't interleave with rendering. + select() menus. It is safe to call from any thread. + + If no prompt is running, or this is called on the prompt's event loop, then the title is + written and flushed before this returns, and output errors propagate to the caller. + + If another thread calls this while a prompt is running, then the title is written on the + prompt's event 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. + Output errors are ignored so they can't disrupt the prompt. :param title: the new window title """ @@ -5917,7 +5932,7 @@ def set_window_title(self, title: str) -> None: if not on_loop: try: - loop.call_soon_threadsafe(self._flush_pending_window_title) + 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 @@ -5930,9 +5945,14 @@ def set_window_title(self, title: str) -> None: 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.""" - apps = [self.active_session.app] - if self._select_app_session is not None and self._select_app_session.app is not None: - apps.append(self._select_app_session.app) + 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 @@ -5940,6 +5960,13 @@ def _get_prompt_loop(self) -> asyncio.AbstractEventLoop | None: return loop return None + 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 self._window_title_fallback_output is not None: + return self._window_title_fallback_output + 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: @@ -5948,10 +5975,18 @@ def _flush_pending_window_title(self) -> None: return self._pending_window_title = None - output = self.main_session.output + output = self._get_window_title_output() output.set_title(title) output.flush() + 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. + """ + with contextlib.suppress(OSError): + self._flush_pending_window_title() + 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 40bed13de..c960537f1 100644 --- a/docs/features/prompt.md +++ b/docs/features/prompt.md @@ -58,11 +58,17 @@ Call this method on your application instance, for example `app.set_window_title 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 is safe to call from any thread, and the title is flushed right away. When another thread -calls it while a prompt or `select()` menu is running, the title is written on that prompt's event -loop so it doesn't interleave with rendering. In that case, output errors are reported by the -prompt's event loop. Otherwise, errors such as a closed terminal propagate to the caller. -Non-interactive sessions with a dummy output do not emit a title. +The method is safe to call from any thread. If no prompt is running, or it is called on the prompt's +event loop, then the title is flushed before the method returns, and output errors such as a closed +terminal propagate to the caller. + +When another thread calls it while a prompt or `select()` menu is running, the title is written on +that prompt's event 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. Output errors are ignored in that case so they can't disrupt the prompt. + +When stdin isn't a terminal but stdout is, titles are still written to stdout. When stdout isn't a +terminal, 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) diff --git a/tests/test_window_title.py b/tests/test_window_title.py index 816aa4668..37b9d4993 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -346,19 +346,21 @@ def after_render(_sender): worker.join(timeout=5) assert not worker.is_alive() + # The loop's exception handler never hears of the error, so it can't disrupt the prompt. + assert not loop_errors if from_worker: - # The title was written by an event loop callback, so the loop's exception handler reports the error. + # The title was written by an event loop callback, which ignores output errors. assert not caught - assert len(loop_errors) == 1 - assert str(loop_errors[0]["exception"]) == "terminal closed" + assert app._pending_window_title is None else: assert caught == ["terminal closed"] - assert not loop_errors -def test_title_is_written_when_the_prompt_finishes_before_running_the_callback(monkeypatch): +@pytest.mark.parametrize("failing", [False, True]) +def test_title_is_written_when_the_prompt_finishes_before_running_the_callback(monkeypatch, failing): stream = TtyStream() - output = title_output(stream) + 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): @@ -372,13 +374,13 @@ def update_title(): # 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: + 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("Finished") + app.set_window_title(title) schedule(lambda: prompt_app.exit(result="done")) def after_render(_sender): @@ -397,9 +399,96 @@ def after_render(_sender): assert len(dropped) == 1 assert app._pending_window_title is None - # The prompt's cleanup wrote the title after the prompt finished rendering. - assert stream.getvalue().endswith("\x1b]2;Finished\x07") - assert output.title_loops["Finished"] 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_does_not_deadlock(monkeypatch): + stream = TtyStream() + errors = [] + + class ReentrantOutput(TitleOutput): + def set_title(self, title): + if title == "Outer": + # Like a signal handler which sets the title while the main thread is writing one. + app.set_window_title("Inner") + super().set_title(title) + + with create_pipe_input() as pipe_input: + app = make_app(monkeypatch, pipe_input, ReentrantOutput(stream)) + + def set_outer_title(): + try: + app.set_window_title("Outer") + except BaseException as error: # noqa: BLE001 + errors.append(error) + + worker = threading.Thread(target=set_outer_title, daemon=True) + worker.start() + worker.join(timeout=5) + assert not worker.is_alive() + + assert not errors + assert stream.getvalue() == "\x1b]2;Inner\x07\x1b]2;Outer\x07" + + +def test_title_falls_back_to_a_terminal_stdout_when_stdin_is_not_a_terminal(monkeypatch): + stream = TtyStream() + output = title_output(stream) + monkeypatch.setattr("cmd2.cmd2.create_output", lambda **_kwargs: output) + app = cmd2.Cmd(stdin=io.StringIO(), stdout=stream, allow_cli_args=False) + assert isinstance(app.main_session.output, DummyOutput) + app.set_window_title("Piped input") + assert stream.getvalue() == "\x1b]2;Piped input\x07" + + # 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_fallback_is_skipped_when_the_terminal_output_cannot_be_created(monkeypatch): + def no_console(**_kwargs): + raise ValueError("no console") + + monkeypatch.setattr("cmd2.cmd2.create_output", no_console) + app = cmd2.Cmd(stdin=io.StringIO(), stdout=TtyStream(), allow_cli_args=False) + assert app._window_title_fallback_output is None + app.set_window_title("Nowhere") + assert app.stdout.getvalue() == "" def test_title_is_written_directly_when_the_prompt_loop_is_closed(monkeypatch): From a33ccdd14ba6cc8d963b1e710cecf6c93e506a4e Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 22:45:20 -0400 Subject: [PATCH 08/12] Fix window title edge cases found in second review - When a signal handler sets the title while this thread is writing one, only record it. The outer call writes pending titles until none remain, so the latest title wins and writes never nest in the same stream. - Ignore ValueError from a closed stream as well as OSError when writing a title quietly. - Write a title left pending by a prompt after patch_stdout() exits, so it can't race with its writer thread. - Create the fallback terminal output the first time a title needs it, from the stdout captured when the app was created. This also covers a main_session replaced later by one with a DummyOutput, and titles set while output is redirected. - Share the list of terminal initialization errors between the main session and the title fallback. --- cmd2/cmd2.py | 80 +++++++++++++++++--------- tests/test_window_title.py | 113 +++++++++++++++++++++++++++++++++++-- 2 files changed, 160 insertions(+), 33 deletions(-) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 29ef6bce7..ce3054b69 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -221,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.""" @@ -691,16 +696,17 @@ def __init__( 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. + # 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 - # When stdin isn't a terminal, the main session uses a DummyOutput. If stdout is a terminal, then - # window titles are still written to it. + # When stdin isn't a terminal, the main session uses a DummyOutput. If stdout was a terminal when this app was + # created, then window titles are still written to it, including while a command's output is redirected. + # Its output is created the first time a title needs it. + self._window_title_fallback_stream: TextIO | None = self.stdout if self.stdout.isatty() else None self._window_title_fallback_output: Output | None = None - if isinstance(self.main_session.output, DummyOutput) and self.stdout.isatty(): - with contextlib.suppress(NoConsoleScreenBufferError, AttributeError, ValueError): - self._window_title_fallback_output = create_output(stdout=self.stdout) # 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 @@ -854,7 +860,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. @@ -3675,15 +3681,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 - self._flush_pending_window_title_quietly() + 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() @@ -5963,28 +5973,44 @@ def _get_prompt_loop(self) -> asyncio.AbstractEventLoop | None: 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 self._window_title_fallback_output is not None: - return self._window_title_fallback_output - return output + if not isinstance(output, DummyOutput): + return output + + if self._window_title_fallback_output is None and self._window_title_fallback_stream is not None: + try: + self._window_title_fallback_output = create_output(stdout=self._window_title_fallback_stream) + except _TERMINAL_INIT_ERRORS: + # Don't try again for every title. + self._window_title_fallback_stream = None + + return self._window_title_fallback_output or 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: - title = self._pending_window_title - if title is None: + 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._pending_window_title = None - output = self._get_window_title_output() - output.set_title(title) - output.flush() + 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. + An error here would otherwise disrupt the prompt or replace its result. Writing to a closed stream raises + ValueError rather than OSError. """ - with contextlib.suppress(OSError): + with contextlib.suppress(OSError, ValueError): self._flush_pending_window_title() def enable_command(self, command: str) -> None: diff --git a/tests/test_window_title.py b/tests/test_window_title.py index 37b9d4993..6d83a2fae 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -1,6 +1,7 @@ """Window title behavior through real prompt-toolkit output and applications.""" import asyncio +import contextlib import contextvars import io import sys @@ -433,17 +434,31 @@ def test_prompt_loop_ignores_a_select_session_without_an_app(monkeypatch): assert app._get_prompt_loop() is None -def test_title_set_while_writing_a_title_does_not_deadlock(monkeypatch): +def test_title_set_while_writing_a_title_is_written_next(monkeypatch): stream = TtyStream() errors = [] 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 the main thread is writing one. + # 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)) @@ -453,23 +468,88 @@ def set_outer_title(): except BaseException as error: # noqa: BLE001 errors.append(error) + # Run in a thread so a deadlock fails the test instead of hanging it. worker = threading.Thread(target=set_outer_title, daemon=True) worker.start() worker.join(timeout=5) assert not worker.is_alive() assert not errors - assert stream.getvalue() == "\x1b]2;Inner\x07\x1b]2;Outer\x07" + # 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) - monkeypatch.setattr("cmd2.cmd2.create_output", lambda **_kwargs: output) + 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") - assert stream.getvalue() == "\x1b]2;Piped input\x07" + 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() @@ -480,16 +560,37 @@ def test_title_falls_back_to_a_terminal_stdout_when_stdin_is_not_a_terminal(monk 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)) + assert app._window_title_fallback_output is None + + 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) - assert app._window_title_fallback_output is None 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() From dcdfdc58eff1cd2790d815788733a809cc37717f Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:00:38 -0400 Subject: [PATCH 09/12] Schedule every title on a running prompt's loop and check stdout lazily - Treat a prompt as running while its application's loop is set, which covers prompt-toolkit's teardown after it clears is_running but can still write to the terminal. - Schedule the title on the prompt's loop even when called on that loop, so a signal handler can't nest a title write inside prompt-toolkit's own flush. - Check whether the fallback stdout is a terminal the first time a title needs it, so constructing a Cmd with a stdout that lacks isatty() or is closed works as it did before. - Describe the stdout fallback in the set_window_title() docstring and the prompt docs. - Run test workers through futures so their exceptions reach the test without a blind except and noqa. --- CHANGELOG.md | 8 +- cmd2/cmd2.py | 65 ++++++++-------- docs/features/prompt.md | 18 ++--- tests/test_window_title.py | 149 ++++++++++++++++++++++--------------- 4 files changed, 134 insertions(+), 106 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 301c17b45..1aae5e8c2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,10 +35,10 @@ 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. When another thread - sets it 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 when stdin isn't a terminal but stdout is. + 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 when stdin isn't a terminal but stdout is. ## 4.3.1 (October 24, 2026) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index ce3054b69..9a9e7be49 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -702,10 +702,10 @@ def __init__( self._pending_window_title: str | None = None self._writing_window_title = False - # When stdin isn't a terminal, the main session uses a DummyOutput. If stdout was a terminal when this app was - # created, then window titles are still written to it, including while a command's output is redirected. - # Its output is created the first time a title needs it. - self._window_title_fallback_stream: TextIO | None = self.stdout if self.stdout.isatty() else None + # When stdin isn't a terminal, the main session uses a DummyOutput. If the stdout this app was created with is + # a terminal, then window titles are still written to it, including while a command's output is redirected. + # Whether it's a terminal is checked, and its output created, the first time a title needs it. + self._window_title_fallback_stream: TextIO | None = self.stdout 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. @@ -5919,14 +5919,16 @@ def set_window_title(self, title: str) -> None: """Set the title of this application's terminal. The title is written to the same output as the main prompt, custom input prompts, and - select() menus. It is safe to call from any thread. + select() menus. When stdin isn't a terminal, those use a dummy output, so the title is + written to the stdout this app was created with instead, if that is a terminal. This method + is safe to call from any thread. - If no prompt is running, or this is called on the prompt's event loop, then the title is - written and flushed before this returns, and output errors propagate to the caller. + If no prompt is running, then the title is written and flushed before this returns, and + output errors propagate to the caller. - If another thread calls this while a prompt is running, then the title is written on the - prompt's event 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. + 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 never interleaves 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. Output errors are ignored so they can't disrupt the prompt. :param title: the new window title @@ -5936,25 +5938,23 @@ def set_window_title(self, title: str) -> None: loop = self._get_prompt_loop() if loop is not None: try: - on_loop = asyncio.get_running_loop() is loop + loop.call_soon_threadsafe(self._flush_pending_window_title_quietly) except RuntimeError: - on_loop = False - - if not on_loop: - 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. - return + # 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. + return 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.""" + """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. @@ -5966,7 +5966,7 @@ def _get_prompt_loop(self) -> asyncio.AbstractEventLoop | None: for app in apps: loop = app.loop - if app.is_running and loop is not None: + if loop is not None: return loop return None @@ -5976,12 +5976,15 @@ def _get_window_title_output(self) -> Output: if not isinstance(output, DummyOutput): return output - if self._window_title_fallback_output is None and self._window_title_fallback_stream is not None: - try: - self._window_title_fallback_output = create_output(stdout=self._window_title_fallback_stream) - except _TERMINAL_INIT_ERRORS: - # Don't try again for every title. - self._window_title_fallback_stream = None + stream = self._window_title_fallback_stream + if self._window_title_fallback_output is None and stream is not None: + # A stream without isatty(), or a closed one, raises AttributeError or ValueError here. + with contextlib.suppress(*_TERMINAL_INIT_ERRORS): + if stream.isatty(): + self._window_title_fallback_output = create_output(stdout=stream) + + # Whatever the outcome, don't check again for every title. + self._window_title_fallback_stream = None return self._window_title_fallback_output or output diff --git a/docs/features/prompt.md b/docs/features/prompt.md index c960537f1..a59ee348d 100644 --- a/docs/features/prompt.md +++ b/docs/features/prompt.md @@ -58,17 +58,17 @@ Call this method on your application instance, for example `app.set_window_title 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 is safe to call from any thread. If no prompt is running, or it is called on the prompt's -event loop, then the title is flushed before the method returns, and output errors such as a closed -terminal propagate to the caller. +The method is safe to call 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. -When another thread calls it while a prompt or `select()` menu is running, the title is written on -that prompt's event 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. Output errors are ignored in that case so they can't disrupt the prompt. +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 never interleaves 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. Output errors are ignored in that case so they can't disrupt the prompt. -When stdin isn't a terminal but stdout is, titles are still written to stdout. When stdout isn't a -terminal, no title is emitted. +When stdin isn't a terminal, prompts use a dummy output, so titles are written to the stdout the +application was created with instead, if that is a terminal. They still go there while a command's +output is redirected. When that stdout isn't a terminal, 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) diff --git a/tests/test_window_title.py b/tests/test_window_title.py index 6d83a2fae..390f83fb0 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -56,6 +56,18 @@ 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() @@ -87,7 +99,6 @@ def test_title_before_during_and_after_a_real_prompt(monkeypatch, from_worker, c foreign = TtyStream() output = title_output(stream) observed = [] - errors = [] workers = [] prompt_loops = [] run = Application.run @@ -100,8 +111,6 @@ 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")) @@ -112,9 +121,7 @@ def after_render(prompt_app): sent = True prompt_loops.append(prompt_app.loop) if from_worker: - worker = threading.Thread(target=contextvars.Context().run, args=(update_title, prompt_app), daemon=True) - workers.append(worker) - worker.start() + workers.append(start_worker(update_title, prompt_app)) else: update_title(prompt_app) @@ -129,17 +136,15 @@ def run_prompt(prompt_app, *args, **kwargs): 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() + worker.result(timeout=5) app.set_window_title("After") assert result == "done" - assert not errors - # A worker's title is written asynchronously on the prompt's event loop, so it may or may not be in the stream when - # set_window_title() returns. Only a call on the loop thread is guaranteed to have written it by then. + # 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 == [True] + 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 @@ -154,7 +159,6 @@ def test_worker_titles_during_a_real_select_and_after_cleanup(monkeypatch, inter stream = TtyStream() foreign = TtyStream() output = title_output(stream) - errors = [] menu_loops = [] run = Application.run @@ -165,8 +169,6 @@ def run_menu(menu, *args, **kwargs): def update_title(): try: app.set_window_title("Menu") - except BaseException as error: # noqa: BLE001 - errors.append(error) finally: if interrupt: menu.loop.call_soon_threadsafe(lambda: menu.exit(exception=KeyboardInterrupt)) @@ -178,17 +180,14 @@ def after_render(_sender): if not sent: sent = True menu_loops.append(menu.loop) - worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True) - workers.append(worker) - worker.start() + 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.join(timeout=5) - assert not worker.is_alive() + 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) @@ -201,7 +200,6 @@ def after_render(_sender): assert app._select_app_session is None app.set_window_title("Finished") - assert not errors assert output.title_loops["Menu"] is menu_loops[0] assert output.title_loops["Finished"] is None assert "\x1b]2;Menu\x07" in stream.getvalue() @@ -335,26 +333,20 @@ def after_render(_sender): 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() + 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.join(timeout=5) - assert not worker.is_alive() + worker.result(timeout=5) - # The loop's exception handler never hears of the error, so it can't disrupt the prompt. + # 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 - if from_worker: - # The title was written by an event loop callback, which ignores output errors. - assert not caught - assert app._pending_window_title is None - else: - assert caught == ["terminal closed"] + assert app._pending_window_title is None @pytest.mark.parametrize("failing", [False, True]) @@ -388,15 +380,12 @@ 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() + 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.join(timeout=5) - assert not worker.is_alive() + worker.result(timeout=5) assert len(dropped) == 1 assert app._pending_window_title is None @@ -436,7 +425,6 @@ def test_prompt_loop_ignores_a_select_session_without_an_app(monkeypatch): def test_title_set_while_writing_a_title_is_written_next(monkeypatch): stream = TtyStream() - errors = [] class ReentrantOutput(TitleOutput): flushing = False @@ -462,19 +450,9 @@ def flush(self): with create_pipe_input() as pipe_input: app = make_app(monkeypatch, pipe_input, ReentrantOutput(stream)) - def set_outer_title(): - try: - app.set_window_title("Outer") - except BaseException as error: # noqa: BLE001 - errors.append(error) - # Run in a thread so a deadlock fails the test instead of hanging it. - worker = threading.Thread(target=set_outer_title, daemon=True) - worker.start() - worker.join(timeout=5) - assert not worker.is_alive() + start_worker(app.set_window_title, "Outer").result(timeout=5) - assert not errors # 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 @@ -599,15 +577,69 @@ def test_title_is_written_directly_when_the_prompt_loop_is_closed(monkeypatch): loop.close() with create_pipe_input() as pipe_input: app = make_app(monkeypatch, pipe_input, output) - prompt_app = app.main_session.app - monkeypatch.setattr(prompt_app, "_is_running", True) - monkeypatch.setattr(prompt_app, "loop", loop) + 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 + + +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") + assert app._window_title_fallback_output is None + assert app._window_title_fallback_stream 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") @@ -636,7 +668,6 @@ def test_replacing_main_session_keeps_titles_on_the_new_output(): stream = TtyStream() foreign = TtyStream() output = title_output(stream) - errors = [] workers = [] prompt_loops = [] with create_pipe_input() as pipe_input, create_app_session(input=DummyInput(), output=title_output(foreign)): @@ -649,8 +680,6 @@ def test_replacing_main_session_keeps_titles_on_the_new_output(): def update_title(): try: app.set_window_title("Replacement prompt") - 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")) @@ -659,18 +688,14 @@ def after_render(_sender): if not sent: sent = True prompt_loops.append(app.main_session.app.loop) - worker = threading.Thread(target=contextvars.Context().run, args=(update_title,), daemon=True) - workers.append(worker) - worker.start() + 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.join(timeout=5) - assert not worker.is_alive() + worker.result(timeout=5) app.set_window_title("After") - assert not errors 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() From 57081a2c38cf7ecf7f979ca390452d124a126f19 Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:18:55 -0400 Subject: [PATCH 10/12] Coalesce queued title callbacks and clarify when titles use the stdout fallback - While a title callback is queued on a prompt's loop, later titles only replace the pending title, so a burst of updates schedules one callback. The callback and prompt cleanup clear the queued loop. - Keep the stdout this app was created with in its own field, and track whether its fallback output has been checked with a flag instead of clearing that field. A cached_property doesn't work here, because inspect.getmembers() evaluates properties while a Cmd is constructed. - Describe all the cases where prompts use a dummy output, and note that a direct write from another thread isn't synchronized with other writers to the terminal. --- CHANGELOG.md | 3 +- cmd2/cmd2.py | 75 ++++++++++++++++++++++++-------------- docs/features/prompt.md | 18 +++++---- tests/test_window_title.py | 69 ++++++++++++++++++++++++++++++++++- 4 files changed, 128 insertions(+), 37 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1aae5e8c2..2bc9b730b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,7 +38,8 @@ 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 when stdin isn't a terminal but stdout is. + 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 9a9e7be49..147553ca6 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -702,10 +702,14 @@ def __init__( self._pending_window_title: str | None = None self._writing_window_title = False - # When stdin isn't a terminal, the main session uses a DummyOutput. If the stdout this app was created with is - # a terminal, then window titles are still written to it, including while a command's output is redirected. - # Whether it's a terminal is checked, and its output created, the first time a title needs it. - self._window_title_fallback_stream: TextIO | None = self.stdout + # 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. @@ -5918,18 +5922,21 @@ def add_alert( def set_window_title(self, title: str) -> None: """Set the title of this application's terminal. - The title is written to the same output as the main prompt, custom input prompts, and - select() menus. When stdin isn't a terminal, those use a dummy output, so the title is - written to the stdout this app was created with instead, if that is a terminal. This method - is safe to call from any thread. + 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. - If no prompt is running, then the title is written and flushed before this returns, and - output errors propagate to the caller. + 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 never interleaves 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. - Output errors are ignored so they can't disrupt the prompt. + 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 """ @@ -5937,6 +5944,10 @@ def set_window_title(self, title: str) -> None: 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: @@ -5945,6 +5956,7 @@ def set_window_title(self, title: str) -> None: 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 self._flush_pending_window_title() @@ -5970,23 +5982,28 @@ def _get_prompt_loop(self) -> asyncio.AbstractEventLoop | None: return loop return None - def _get_window_title_output(self) -> Output: - """Return the output which window titles are written to.""" - output = self.main_session.output - if not isinstance(output, DummyOutput): - return output + 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 - stream = self._window_title_fallback_stream - if self._window_title_fallback_output is None and stream is not None: # A stream without isatty(), or a closed one, raises AttributeError or ValueError here. with contextlib.suppress(*_TERMINAL_INIT_ERRORS): - if stream.isatty(): - self._window_title_fallback_output = create_output(stdout=stream) + if self._window_title_stdout.isatty(): + self._window_title_fallback_output = create_output(stdout=self._window_title_stdout) - # Whatever the outcome, don't check again for every title. - self._window_title_fallback_stream = None + return self._window_title_fallback_output - return self._window_title_fallback_output or 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.""" @@ -6013,8 +6030,12 @@ def _flush_pending_window_title_quietly(self) -> None: An error here would otherwise disrupt the prompt or replace its result. Writing to a closed stream raises ValueError rather than OSError. """ - with contextlib.suppress(OSError, ValueError): - self._flush_pending_window_title() + 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/features/prompt.md b/docs/features/prompt.md index a59ee348d..59aeeed05 100644 --- a/docs/features/prompt.md +++ b/docs/features/prompt.md @@ -58,17 +58,21 @@ Call this method on your application instance, for example `app.set_window_title 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 is safe to call 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. +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 never interleaves with rendering. The title appears +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. Output errors are ignored in that case so they can't disrupt the prompt. +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. -When stdin isn't a terminal, prompts use a dummy output, so titles are written to the stdout the -application was created with instead, if that is a terminal. They still go there while a command's -output is redirected. When that stdout isn't a terminal, no title is emitted. +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) diff --git a/tests/test_window_title.py b/tests/test_window_title.py index 390f83fb0..e99108820 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -543,7 +543,8 @@ def test_title_falls_back_when_a_replacement_session_has_no_terminal_output(monk fallback = TtyStream() with create_pipe_input() as pipe_input: app = make_app(monkeypatch, pipe_input, title_output(stream)) - assert app._window_title_fallback_output is None + # 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)) @@ -607,6 +608,69 @@ def test_title_is_scheduled_while_a_finishing_prompt_still_has_its_loop(monkeypa 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: + with create_pipe_input() as pipe_input: + 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().""" @@ -634,8 +698,9 @@ def test_stdout_which_cannot_report_a_terminal_does_not_break_titles(closed): 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 - assert app._window_title_fallback_stream is None if not closed: assert stdout.written == [] From e72ceb558328b5cd4211d27605c76a7897b3d5b0 Mon Sep 17 00:00:00 2001 From: Todd Leonhardt <2812720+tleonhardt@users.noreply.github.com> Date: Wed, 28 Oct 2026 23:24:36 -0400 Subject: [PATCH 11/12] Give the prompt cleanup test a VT100 output for patch_stdout() On a Windows CI runner the default app session output is a Win32Output, which raises NoConsoleScreenBufferError because there's no console. Run the test inside create_app_session() like the other prompt tests. --- tests/test_window_title.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/test_window_title.py b/tests/test_window_title.py index e99108820..f1f06d4fd 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -650,7 +650,8 @@ def test_prompt_cleanup_clears_a_callback_its_loop_never_ran(monkeypatch): output = title_output(stream) stopped_loop = asyncio.new_event_loop() try: - with create_pipe_input() as pipe_input: + # 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 From 4472a0afb50539b589733ada878db22350397c61 Mon Sep 17 00:00:00 2001 From: KennySimpson <70093673+KennyMcSimpson@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:35:07 +0800 Subject: [PATCH 12/12] Keep synchronous window title errors with their caller (#1793) --- cmd2/cmd2.py | 4 ++- tests/test_window_title.py | 60 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 1 deletion(-) diff --git a/cmd2/cmd2.py b/cmd2/cmd2.py index 147553ca6..28ebf48df 100644 --- a/cmd2/cmd2.py +++ b/cmd2/cmd2.py @@ -5959,7 +5959,9 @@ def set_window_title(self, title: str) -> None: self._window_title_callback_loop = loop return - self._flush_pending_window_title() + # 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. diff --git a/tests/test_window_title.py b/tests/test_window_title.py index f1f06d4fd..1b0d598fc 100644 --- a/tests/test_window_title.py +++ b/tests/test_window_title.py @@ -307,6 +307,66 @@ def test_title_output_errors_reach_the_caller_without_a_prompt(monkeypatch, oper 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):