Address five review findings on the edge-case fixes

Abandoning the reservation while the command display was paused for a
guest left the display on the empty reserved layout with no bridge behind
it, so the native toolbar never appeared for the rest of the command. The
layout swap now happens as soon as the reservation stops, since it needs
no UI loop; only a running display's routing and redraw are scheduled on
its loop.

The unfinished-output verdict that starts the next prompt on a fresh line
is now dropped when a guest takes the terminal or the screen is erased,
since the cursor is no longer where the output left it, and it ignores
trailing control sequences -- a reset after the newline no longer counts
as text on a new line, and a write made only of control says nothing.

The size-poll test names the upstream coroutine it depends on. Starting
the reserved toolbar on an unqualified backend binds nothing, rather than
an adapter and bridge that could never reserve.

Two small tests cover the defensive returns in the legacy fallback, so
command_toolbar.py is back at its previous coverage.

Validation: 2624 passed, 6 skipped with coverage, twice; five mutations
each fail their test; harness acceptance and dynamic gates PASS at 12, 24
and 40 rows, 23/23 observer controls. make check, make test and
make docs-test passed.

Author: tleonhardt

CHANGELOG.md

@@ -4,9 +4,12 @@
     - Background output printed above an active reserved prompt through `run_in_terminal()` or
       `patch_stdout()` is preserved when the prompt redraws.
     - Reserved rendering preserves unfinished command output when the main prompt returns by
-      starting the prompt on a fresh line.
+      starting the prompt on a fresh line. Trailing control sequences do not count as unfinished
+      output, and a line another program finished during a terminal handoff is not given an extra
+      blank line.
     - Falling back from reserved rendering during a command restores the native toolbar layout and
-      stdout proxy, so a recovered toolbar remains visible during the command.
+      stdout proxy, so a recovered toolbar remains visible during the command, including when the
+      fallback happens while the command has handed the terminal to another program.
     - Resuming the reserved command display after a terminal handoff preserves the cursor column of
       unfinished guest output, so later command output continues the same line.
     - A reserved toolbar started in a terminal below the minimum height now activates when the

cmd2/command_toolbar.py

@@ -229,9 +229,9 @@ def __init__(self, cmd: "Cmd") -> None:
         # Legacy rendering still needs the filler to push its scrolling toolbar to the bottom.
         reserved = cmd.reserved_toolbar
         if reserved is not None and reserved.bridge is not None:
-            self._layout = Layout(HSplit([Window(height=0)]))
+            self._layout = self._reserved_layout()
         else:
-            self._layout = Layout(HSplit([Window(height=0), Window(), self.toolbar]))
+            self._layout = self._legacy_layout()
         self._display_stack: contextlib.ExitStack | None = None
         bindings = KeyBindings()
 
@@ -398,15 +398,36 @@ def _install_legacy_proxy(self) -> None:
                 stream.serializer = None
                 stream.proxy = proxy
 
+    @staticmethod
+    def _reserved_layout() -> Layout:
+        """Build the command display's layout for reserved rendering: nothing of its own."""
+        return Layout(HSplit([Window(height=0)]))
+
+    def _legacy_layout(self) -> Layout:
+        """Build the command display's layout for legacy rendering: a filler and the toolbar."""
+        return Layout(HSplit([Window(height=0), Window(), self.toolbar]))
+
     def _reservation_stopped(self) -> None:
-        """Restore legacy layout and routing on the UI loop after safe physical release."""
+        """Fall back to the legacy display once the reservation has been physically released.
+
+        The layout swap happens here, unconditionally: it needs no UI loop, and the next
+        resume reads it. The reservation can stop while the display is paused for a guest,
+        when there is no running loop to switch the live layout on -- swapping only from the
+        loop callback would bring the display back as the empty reserved layout with no
+        bridge behind it, and the native toolbar would never appear for the rest of the
+        command. Routing and the redraw of a display that *is* running need its loop.
+        """
+        previous_layout = self._layout
+        self._layout = self._legacy_layout()
         if self.app.loop is not None and self.app.is_running:
-            self.app.loop.call_soon_threadsafe(self._restore_legacy_display)
+            self.app.loop.call_soon_threadsafe(self._restore_legacy_display, previous_layout)
 
-    def _restore_legacy_display(self) -> None:
-        """Replace the empty reserved display once its bridge has been removed."""
-        previous_layout = self._layout
-        self._layout = Layout(HSplit([Window(height=0), Window(), self.toolbar]))
+    def _restore_legacy_display(self, previous_layout: Layout) -> None:
+        """Switch a running display over to legacy routing and layout, on its own loop.
+
+        :param previous_layout: the reserved layout the display was started with, replaced on
+            the application only if it is still the one in use
+        """
         if self._pausing or not self.app.is_running or self.app.is_done:
             return
         self._install_legacy_proxy()

cmd2/prompt_toolkit_bridge.py

@@ -24,6 +24,7 @@
 and :mod:`tests.test_prompt_toolkit_bridge` holds it to that version by name.
 """
 
+import re
 from collections import deque
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any
@@ -43,6 +44,13 @@
     from .terminal_display import TerminalDisplay
 
 
+#: Trailing control that moves nothing the user can see: carriage returns, CSI sequences such
+#: as an SGR reset, and OSC sequences such as a window title. Stripped before deciding whether
+#: output ended on a fresh line, so a reset emitted after the newline does not count as text
+#: on a new line, and a write made only of control says nothing about the line at all.
+_TRAILING_CONTROL = re.compile(r"(?:\r|\x1b\[[0-?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\))+\Z")
+
+
 class ReservedModeFailureError(RuntimeError):
     """Raised when reserved rendering cannot continue safely.
 
@@ -263,7 +271,9 @@ def note_managed_write(self, prompt_anchor: int | None = None, *, data: str | No
         """
         self._terminal_generation += 1
         if data:
-            self._unfinished_command_output = not data.rstrip("\r").endswith("\n")
+            visible = _TRAILING_CONTROL.sub("", data)
+            if visible:
+                self._unfinished_command_output = not visible.endswith("\n")
         self._prompt_anchor = prompt_anchor
         # Only the frame in flight goes; it was prepared against the cursor and content this
         # write moved, so committing it would emit a stale frame. The renderer's baseline is
@@ -274,6 +284,22 @@ def note_managed_write(self, prompt_anchor: int | None = None, *, data: str | No
         self._renderer._last_screen = self._committed_screen
         self._request_redraw()
 
+    @property
+    def has_unfinished_command_output(self) -> bool:
+        """Whether the last visible command output stopped part-way through a line."""
+        return self._unfinished_command_output
+
+    def forget_unfinished_command_output(self) -> None:
+        """Drop the verdict about a line in progress: the cursor is no longer where it left it.
+
+        A guest program that took the terminal, or an erase that cleared the screen, has moved
+        the cursor since. The line the verdict described may well be finished by now, and
+        adding a newline for it would push the next prompt down by a blank line. Output the
+        guest itself left unfinished is not seen here -- it bypasses the serializer -- and is
+        the guest's to finish, as it was before the reservation existed.
+        """
+        self._unfinished_command_output = False
+
     def finish_command_output(self) -> None:
         """Start the next prompt on a fresh line if command output left one unfinished.
 
@@ -674,6 +700,8 @@ def _erase_through_bridge(self, leave_alternate_screen: bool = True) -> None:
                 # recovering at the old prompt row and erasing the callback's output.
                 self._prompt_anchor = None
                 self._invalidate_pending_cursor_reports()
+                # The screen below the cursor is gone, and with it any line in progress.
+                self._unfinished_command_output = False
                 self.require_resynchronization("the renderer erased the screen")
 
     def _clear_through_bridge(self) -> None:
@@ -699,6 +727,7 @@ def _clear_through_bridge(self) -> None:
             finally:
                 self._prompt_anchor = None
                 self._invalidate_pending_cursor_reports()
+                self._unfinished_command_output = False
                 self.require_resynchronization("the renderer cleared the screen")
 
     # -- prepare and commit ----------------------------------------------------------------

cmd2/reserved_toolbar.py

@@ -164,6 +164,12 @@ def start(self) -> bool:
             raise RuntimeError("cannot locate the session's bottom toolbar window")
 
         display = TerminalDisplay(app.output, reserved_rows=self._reserved_rows)
+        if not display.terminal.supports_reservation:
+            # Mode selection refuses an unqualified backend before this runs, so this is for
+            # a caller using the class directly. Nothing is bound: a terminal below the floor
+            # keeps a lease and a bridge for the resize that may make a reservation possible,
+            # but an unqualified backend can never reserve, whatever its size.
+            return False
         display.acquire()
 
         self._display = display
@@ -290,6 +296,7 @@ def _invalidate_ownership(self, reason: str) -> None:
             self._painter.invalidate()
         if self._bridge is not None:
             self._bridge.forget_prompt_anchor()
+            self._bridge.forget_unfinished_command_output()
             self._bridge.require_resynchronization(reason)
 
     def refresh(self) -> bool:

tests/test_command_toolbar.py

@@ -1149,3 +1149,30 @@ def test_a_surviving_display_stops_another_from_starting(toolbar_app, expire_sta
             assert app._command_toolbar is None
     finally:
         blocked.set()
+
+
+def test_installing_the_legacy_proxy_twice_keeps_the_first(toolbar_app) -> None:
+    """Falling back to legacy routing on a display that already routes through a proxy must
+    not replace a proxy whose worker is mid-write; the newcomer is closed instead."""
+    app, _pipe, _output = toolbar_app
+    with app._command_toolbar_context():
+        display = app._command_toolbar
+        assert display is not None
+        proxy = display._proxy
+        assert proxy is not None
+        display._install_legacy_proxy()
+        assert display._proxy is proxy
+        assert all(stream.proxy is proxy for stream in display._streams)
+
+
+def test_restoring_the_legacy_display_on_a_stopped_display_changes_nothing(toolbar_app) -> None:
+    """The fallback can be scheduled just before the display stops; by the time it runs
+    there is no display to switch, and it must not install routing into a closed one."""
+    app, _pipe, _output = toolbar_app
+    with app._command_toolbar_context():
+        display = app._command_toolbar
+        assert display is not None
+    layout = app.main_session.app.layout
+    display._restore_legacy_display(display._layout)
+    assert display._proxy is None
+    assert app.main_session.app.layout is layout

tests/test_prompt_toolkit_bridge.py

@@ -714,6 +714,30 @@ def test_managed_output_drops_the_frame_but_keeps_the_baseline(self) -> None:
         assert harness.renderer._last_screen is committed
         assert harness.bridge.needs_resynchronization is False
 
+    def test_output_ending_in_a_control_sequence_after_its_newline_is_finished(self) -> None:
+        """A reset emitted after the newline does not make the line unfinished."""
+        harness = Harness()
+        harness.bridge.note_managed_write(data="line\n\x1b[0m")
+        assert harness.bridge.has_unfinished_command_output is False
+        harness.bridge.note_managed_write(data="line\n\x1b]0;title\x07")
+        assert harness.bridge.has_unfinished_command_output is False
+
+    def test_a_write_of_only_control_leaves_the_verdict_unchanged(self) -> None:
+        """Nothing visible was written, so nothing about the line in progress changed."""
+        harness = Harness()
+        harness.bridge.note_managed_write(data="PARTIAL")
+        assert harness.bridge.has_unfinished_command_output is True
+        harness.bridge.note_managed_write(data="\x1b[0m")
+        assert harness.bridge.has_unfinished_command_output is True
+        harness.bridge.note_managed_write(data="\n")
+        assert harness.bridge.has_unfinished_command_output is False
+
+    def test_a_bare_carriage_return_keeps_the_line_unfinished(self) -> None:
+        harness = Harness()
+        harness.bridge.note_managed_write(data="progress 50%")
+        harness.bridge.note_managed_write(data="\r")
+        assert harness.bridge.has_unfinished_command_output is True
+
     def test_a_managed_write_can_supply_the_new_prompt_origin(self) -> None:
         """The layer that emitted the output is the one that knows where it ended."""
         harness = Harness()
@@ -1165,6 +1189,27 @@ def test_a_render_whose_commit_is_refused_asks_for_another(self) -> None:
             harness.renderer.render(harness.app, harness.app.layout)
         assert harness.bridge.redraw_pending is True
 
+    @pytest.mark.parametrize("operation", ["erase", "clear"])
+    def test_a_real_erase_forgets_unfinished_output(self, operation) -> None:
+        """After the screen is erased there is no line in progress at the cursor."""
+        harness = self.bound()
+        # clear() asks for a cursor report through the event loop, which this harness lacks.
+        harness.renderer.request_absolute_cursor_position = lambda: None  # type: ignore[method-assign]
+        harness.bridge.note_managed_write(data="PARTIAL")
+        assert harness.bridge.has_unfinished_command_output is True
+        with set_app(harness.app):
+            getattr(harness.renderer, operation)()
+        assert harness.bridge.has_unfinished_command_output is False
+
+    def test_a_suppressed_resize_erase_keeps_the_unfinished_verdict(self) -> None:
+        """The suppressed path skips the erase to keep the line, so the verdict stands."""
+        harness = self.bound()
+        harness.bridge.note_managed_write(data="PARTIAL")
+        harness.bridge.set_render_suppressed(True)
+        with set_app(harness.app):
+            harness.renderer.erase()
+        assert harness.bridge.has_unfinished_command_output is True
+
     def test_an_erase_is_emitted_inside_a_transaction(self) -> None:
         harness = self.bound()
         with set_app(harness.app):

tests/test_reserved_terminal.py

@@ -342,6 +342,9 @@ def observe_size():
                     task = asyncio.current_task()
                 except RuntimeError:
                     task = None
+                # Identified by the name of prompt-toolkit's own polling coroutine,
+                # Application._poll_output_size in the qualified 3.0.53. It is upstream
+                # internal, so this is one of the places a version bump has to revisit.
                 if task is not None and task.get_coro().__name__ == "_poll_output_size":
                     polled.set()
                 return size
@@ -423,6 +426,22 @@ def test_partial_output_survives_the_next_main_prompt(self, terminal_harness, li
             assert sum("IMPORTANT PARTIAL" in row for row in visible) == 1
             assert any("TEST> next" in row for row in visible)
 
+    def test_a_handoff_forgets_unfinished_output_so_the_prompt_is_not_pushed_down(self, terminal_harness) -> None:
+        """A partial write followed by a guest that finishes the line: the guest moved the
+        cursor to a fresh line, so the prompt must not add another one on the strength of a
+        verdict about output the guest has since completed."""
+        harness, terminal = terminal_harness
+        with harness.app._reserved_toolbar_context():
+            with harness.app._command_toolbar_context():
+                harness.app.stdout.write("PARTIAL")
+                harness.app.stdout.flush()
+                with harness.app.suspend_bottom_toolbar():
+                    harness.app.stdout.write("done\n")
+                    harness.app.stdout.flush()
+            read_prompt(harness, terminal)
+            assert terminal.screen.display[0].startswith("PARTIALdone")
+            assert terminal.screen.display[1].startswith("TEST> next")
+
     def test_guest_partial_output_survives_command_resume(self, terminal_harness) -> None:
         harness, terminal = terminal_harness
         with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context():
@@ -572,6 +591,28 @@ def fail():
             assert wait_for(lambda: any("legacy output" in row for row in terminal.screen.display))
             assert terminal.screen.margins is None
 
+    def test_abandoning_reservation_while_suspended_restores_the_legacy_display(self, terminal_harness) -> None:
+        """The reservation can stop while the display is paused for a guest. There is no UI
+        loop to switch the live layout on then, but the display must still come back as the
+        legacy one -- filler, native toolbar, proxy -- rather than the empty reserved layout
+        with no bridge behind it."""
+        harness, terminal = terminal_harness
+        with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context():
+            reserved = harness.app.reserved_toolbar
+            display = harness.app._command_toolbar
+            with harness.app.suspend_bottom_toolbar():
+                reserved.stop()
+            assert reserved.bridge is None
+            assert len(display._layout.container.children) == 3
+            assert display._proxy is not None
+            assert all(stream.serializer is None for stream in display._streams)
+            harness.app.main_session.bottom_toolbar = "RECOVERED"
+            display.app.invalidate()
+            assert wait_for(lambda: any(row.startswith("RECOVERED") for row in terminal.screen.display))
+            harness.app.poutput("legacy output")
+            assert wait_for(lambda: any("legacy output" in row for row in terminal.screen.display))
+        assert terminal.screen.margins is None
+
     def test_the_pager_draws_its_content_over_the_reserved_toolbar(self, terminal_harness) -> None:
         harness, terminal = terminal_harness
         with harness.app._reserved_toolbar_context(), harness.app._command_toolbar_context():

tests/test_reserved_toolbar.py

@@ -130,6 +130,30 @@ def test_stopping_twice_does_nothing_the_second_time(self) -> None:
         finally:
             harness.close()
 
+    def test_an_unqualified_backend_binds_nothing(self) -> None:
+        """Capability is exact class identity, so a subclass is not qualified. Below the floor
+        a lease and a bridge are kept for the resize that may make a reservation possible;
+        an unqualified backend can never reserve, so nothing is bound at all."""
+
+        class OtherOutput(Vt100_Output):
+            pass
+
+        stream = TtyStringIO()
+        backend = OtherOutput(stream, lambda: Size(rows=24, columns=80))
+        with create_pipe_input() as pipe:
+            session: PromptSession[str] = PromptSession(input=pipe, output=backend, bottom_toolbar="STATUS")
+            toolbar = ReservedToolbar(session, lambda: session.bottom_toolbar)
+            # The session's own renderer writes cursor-shape sequences on creation; what
+            # matters is that starting and stopping the toolbar adds nothing to them.
+            before = stream.getvalue()
+            assert toolbar.start() is False
+            assert session.app.output is backend
+            assert session.app.renderer.output is backend
+            assert toolbar.bridge is None
+            assert toolbar.is_active is False
+            toolbar.stop()
+            assert stream.getvalue() == before
+
     def test_a_terminal_too_short_to_reserve_keeps_a_resize_bridge(self) -> None:
         """Below the floor the full-size adapter retains a retry path and cleans up normally."""
         harness = Harness(rows=2)