Stage 3 review: refuse the terminal while a display that would not stop still holds it

Disabling future command displays only stopped another one from starting. The
thread that would not stop was still inside the application -- rendering, and
reading the same input -- and everything else went on as though the terminal
were free: startup reported its timeout, the command ran, and the next
suspension released the margins and handed a guest a terminal cmd2 did not own.

The application now keeps a reference to the display that did not let go, and
refuses to pause, prompt or hand over until it does. That covers the paths that
matter through the two context managers every caller already goes through: the
main prompt, nested prompts, external commands, pagers and finalization.

The refusal heals itself. A render callback that finally returns, a subprocess
that finally exits, and the thread ends; the next check finds it gone and stops
refusing, rather than leaving the session broken for a condition that has
passed.

Author: tleonhardt

cmd2/cmd2.py

@@ -561,6 +561,9 @@ def __init__(
         # so that a typo fails where it was written.
         self._bottom_toolbar_mode = validate_toolbar_mode(bottom_toolbar_mode)
         self._reserved_toolbar: ReservedToolbar | None = None
+        # A command display whose thread did not stop when it was asked to. It still owns the
+        # terminal, so nothing may be handed the terminal until it lets go.
+        self._display_holding_terminal: command_toolbar.CommandToolbar | None = None
 
         # Create the main PromptSession
         self.main_session = self._create_main_session(
@@ -2135,6 +2138,28 @@ def get_bottom_toolbar(self) -> AnyFormattedText:
         """
         return None
 
+    def _require_terminal_ownership(self) -> None:
+        """Refuse to use the terminal while a display that would not stop still holds it.
+
+        A display that timed out on shutdown is disabled for the rest of the session, but that
+        only stops another one from starting. Its thread is still inside the application:
+        rendering, and reading the same input. Handing that terminal to a guest, or prompting
+        on it, would put two readers on one device and interleave their output.
+
+        The check heals itself. The thread may finish late -- a render callback that finally
+        returned, a subprocess that finally exited -- and once it has, the terminal is ours
+        again and this stops refusing.
+
+        :raises RuntimeError: while the surviving display still holds the terminal
+        """
+        display = self._display_holding_terminal
+        if display is None:
+            return
+        if not display.thread_is_alive:
+            self._display_holding_terminal = None
+            return
+        raise RuntimeError("the bottom toolbar's display has not released the terminal")
+
     @contextlib.contextmanager
     def suspend_bottom_toolbar(self) -> Iterator[None]:
         """Temporarily hide the command toolbar and give exclusive access to the terminal.
@@ -2149,6 +2174,7 @@ def suspend_bottom_toolbar(self) -> Iterator[None]:
         """
         with self._quiesce_bottom_toolbar():
             reserved = self._reserved_toolbar
+
             if reserved is None:
                 yield
             else:
@@ -2166,6 +2192,7 @@ def _quiesce_bottom_toolbar(self) -> Iterator[None]:
         something else is another, and the ordinary end of a command needs only the first --
         the toolbar has to still be there when the next prompt appears.
         """
+        self._require_terminal_ownership()
         if self._command_toolbar is None:
             yield
         else:

cmd2/command_toolbar.py

@@ -485,11 +485,13 @@ def _abandon_stuck_display(self) -> None:
 
         The refusal is recorded on the application rather than on this object, which the next
         command replaces. The thread outlives it, and one terminal cannot have two input
-        readers.
+        readers -- so the application keeps a reference to the display that did not let go,
+        and refuses to hand the terminal anywhere until it does.
 
         :raises _DisplayStillRunningError: always
         """
         self.cmd._command_toolbar_disabled = True
+        self.cmd._display_holding_terminal = self
         raise _DisplayStillRunningError(f"the bottom toolbar did not stop within {_SHUTDOWN_TIMEOUT} seconds")
 
     def stop(self) -> None:
@@ -503,6 +505,11 @@ def stop(self) -> None:
                 self._stack.close()
                 self._stack = None
 
+    @property
+    def thread_is_alive(self) -> bool:
+        """Whether the display's thread is still running."""
+        return self._thread is not None and self._thread.is_alive()
+
     @property
     def is_active(self) -> bool:
         """Whether the display currently owns the terminal."""

tests/test_command_toolbar.py

@@ -933,3 +933,80 @@ def test_command_toolbar_that_would_not_stop_keeps_the_application(toolbar_app,
             assert display.app.layout is layout
     finally:
         blocked.set()
+
+
+def test_a_surviving_display_blocks_later_handoffs(toolbar_app, monkeypatch) -> None:
+    """Disabling future displays is not enough: the old one still owns the terminal."""
+    app, _, _ = toolbar_app
+    blocked = threading.Event()
+    monkeypatch.setattr(command_toolbar, "_STARTUP_TIMEOUT", 0.2)
+    monkeypatch.setattr(command_toolbar, "_SHUTDOWN_TIMEOUT", 0.2)
+    entered = []
+
+    try:
+        app.main_session.bottom_toolbar = lambda: blocked.wait(timeout=10) or "STATUS"
+        with app._command_toolbar_context():
+            pass
+
+        # The command display is gone as an object, but its thread is not.
+        assert app._command_toolbar is None
+        with pytest.raises(RuntimeError, match="terminal"), app.suspend_bottom_toolbar():
+            entered.append(True)
+        assert entered == []
+    finally:
+        blocked.set()
+
+
+def test_a_surviving_display_blocks_the_prompt(toolbar_app, monkeypatch) -> None:
+    """Two readers on one terminal is not a state to keep prompting in."""
+    app, _, _ = toolbar_app
+    blocked = threading.Event()
+    monkeypatch.setattr(command_toolbar, "_STARTUP_TIMEOUT", 0.2)
+    monkeypatch.setattr(command_toolbar, "_SHUTDOWN_TIMEOUT", 0.2)
+
+    try:
+        app.main_session.bottom_toolbar = lambda: blocked.wait(timeout=10) or "STATUS"
+        with app._command_toolbar_context():
+            pass
+
+        with pytest.raises(RuntimeError, match="terminal"):
+            app._read_raw_input("> ", app.main_session)
+    finally:
+        blocked.set()
+
+
+def test_the_refusal_lifts_when_the_display_finally_exits(toolbar_app, monkeypatch) -> None:
+    """The thread may yet finish, and the session should not stay broken if it does."""
+    app, _, _ = toolbar_app
+    blocked = threading.Event()
+    monkeypatch.setattr(command_toolbar, "_STARTUP_TIMEOUT", 0.2)
+    monkeypatch.setattr(command_toolbar, "_SHUTDOWN_TIMEOUT", 0.2)
+
+    app.main_session.bottom_toolbar = lambda: blocked.wait(timeout=10) or "STATUS"
+    with app._command_toolbar_context():
+        pass
+    surviving = app._display_holding_terminal
+    assert surviving is not None
+
+    blocked.set()
+    surviving._thread.join(timeout=5)
+
+    with app.suspend_bottom_toolbar():
+        pass
+    assert app._display_holding_terminal is None
+
+
+def test_a_surviving_display_stops_another_from_starting(toolbar_app, monkeypatch) -> None:
+    app, _, _ = toolbar_app
+    blocked = threading.Event()
+    monkeypatch.setattr(command_toolbar, "_STARTUP_TIMEOUT", 0.2)
+    monkeypatch.setattr(command_toolbar, "_SHUTDOWN_TIMEOUT", 0.2)
+
+    try:
+        app.main_session.bottom_toolbar = lambda: blocked.wait(timeout=10) or "STATUS"
+        with app._command_toolbar_context():
+            pass
+        with app._command_toolbar_context():
+            assert app._command_toolbar is None
+    finally:
+        blocked.set()

tests/test_reserved_lifecycle.py

@@ -570,3 +570,31 @@ def test_another_session_still_gets_the_terminal_to_itself(self) -> None:
                 assert toolbar.display.is_reserved is True
         finally:
             harness.close()
+
+
+class TestSurvivingDisplay:
+    def test_the_rows_are_not_released_for_a_terminal_we_do_not_own(self) -> None:
+        """A guest cannot be given rows back while another reader still holds the terminal."""
+        harness = Harness(mode="reserved")
+        try:
+            with harness.app._reserved_toolbar_context():
+                toolbar = harness.app.reserved_toolbar
+                assert toolbar is not None
+
+                # Stand in for a display whose thread never stopped.
+                harness.app._display_holding_terminal = StuckDisplay()
+
+                harness.clear()
+                with pytest.raises(RuntimeError, match="terminal"), harness.app.suspend_bottom_toolbar():
+                    pass
+
+                assert toolbar.display.is_reserved is True
+                assert "\x1b[r" not in harness.written()
+        finally:
+            harness.close()
+
+
+class StuckDisplay:
+    """A command display whose thread will not finish."""
+
+    thread_is_alive = True