Preserve Windows VT processing for serialized command output

Author: tleonhardt

cmd2/command_toolbar.py

@@ -493,7 +493,9 @@ def _install_serializers(self) -> bool:
         with self._lock:
             self._serialized = True
             for stream in self._streams:
-                stream.serializer = SerializedTerminalWriter(stream.original, reserved.lock, reserved.bridge)
+                stream.serializer = SerializedTerminalWriter(
+                    stream.original, reserved.lock, reserved.bridge, output=reserved.display.terminal.output
+                )
         # While the command display owns the terminal it has nothing of its own to draw, so its
         # renderer frames are suppressed: emitting one would reserve the usable height and
         # scroll command output off the screen. The toolbar is painted independently.

cmd2/managed_output.py

@@ -29,21 +29,26 @@
 if TYPE_CHECKING:  # pragma: no cover
     from typing import TextIO
 
+    from prompt_toolkit.output import Output
+
 
 class SerializedTerminalWriter:
     """A stream that writes to the terminal under the terminal transaction lock."""
 
-    def __init__(self, stream: "TextIO", lock: TerminalLock, bridge: Any = None) -> None:
+    def __init__(self, stream: "TextIO", lock: TerminalLock, bridge: Any = None, *, output: "Output | None" = None) -> None:
         """Wrap the terminal's own stream.
 
         :param stream: the *original* terminal stream. Never a proxy over it: a physical
             writer that routed back into a proxy would queue its own output behind itself.
         :param lock: the terminal transaction lock every cmd2-controlled writer shares
         :param bridge: the renderer bridge to inform of managed output, if one is active
+        :param output: the qualified physical backend, when writing to a reserved terminal.
+            Its outer flush owns platform setup such as Windows VT console mode.
         """
         self._stream = stream
         self._lock = lock
         self.bridge = bridge
+        self._output = output
 
     @property
     def stream(self) -> "TextIO":
@@ -58,11 +63,15 @@ def write(self, data: str) -> int:
         """
         with self._lock.transaction("managed write"):
             try:
-                written = self._stream.write(data)
+                if self._output is None:
+                    written = self._stream.write(data)
+                else:
+                    self._output.write_raw(data)
+                    written = len(data)
                 # Flushed before the lock is given up. Left buffered, this output would reach
                 # the terminal after whatever paints next, which is the ordering the
                 # transaction is supposed to establish.
-                self._stream.flush()
+                self.flush()
             finally:
                 # Recorded whether or not the write succeeded, and still inside the
                 # transaction. The bridge drops any frame prepared against the cursor this
@@ -81,7 +90,10 @@ def flush(self) -> None:
         nothing either. The write that produced the buffered output already said so.
         """
         with self._lock.transaction("managed flush"):
-            self._stream.flush()
+            if self._output is None:
+                self._stream.flush()
+            else:
+                self._output.flush()
 
     def __getattr__(self, name: str) -> Any:
         """Delegate file attributes to the terminal stream.

tests/test_managed_output.py

@@ -12,6 +12,9 @@
 from typing import Any, Self
 
 import pytest
+from prompt_toolkit.data_structures import Size
+from prompt_toolkit.output import DummyOutput
+from prompt_toolkit.output.vt100 import Vt100_Output
 
 from cmd2.command_toolbar import ToolbarStream
 from cmd2.managed_output import SerializedTerminalWriter
@@ -99,6 +102,59 @@ def test_file_attributes_come_from_the_stream(self) -> None:
         assert writer.readable() == stream.readable()
 
 
+class TestBackendWriting:
+    @pytest.mark.parametrize("fail_flush", [False, True])
+    def test_outer_backend_flush_owns_vt_mode_inside_the_transaction(self, fail_flush) -> None:
+        """Model Windows' outer backend on every platform: its inner VT writer alone does
+        not enable console processing, and restoration must also happen on failure."""
+        mode = 0
+        observed = []
+
+        class Console(RecordingStream):
+            def write(self, text: str) -> int:
+                observed.append((mode, current_transaction()))
+                return super().write(text)
+
+            def flush(self) -> None:
+                super().flush()
+                if fail_flush:
+                    raise RuntimeError("console flush failed")
+
+        console = Console()
+
+        class WindowsLikeOutput(DummyOutput):
+            def __init__(self) -> None:
+                self.vt100_output = Vt100_Output(console, lambda: Size(24, 80))
+
+            def write_raw(self, data: str) -> None:
+                self.vt100_output.write_raw(data)
+
+            def flush(self) -> None:
+                nonlocal mode
+                original_mode = mode
+                mode = 4
+                try:
+                    self.vt100_output.flush()
+                finally:
+                    mode = original_mode
+
+        bridge = RecordingBridge()
+        writer = SerializedTerminalWriter(console, TerminalLock(), bridge, output=WindowsLikeOutput())
+        data = "\x1b[31mred\x1b[0m\n"
+        if fail_flush:
+            with pytest.raises(RuntimeError, match="console flush failed"):
+                writer.write(data)
+        else:
+            assert writer.write(data) == len(data)
+            writer.flush()
+        assert console.getvalue() == data
+        assert mode == 0
+        assert observed
+        assert all(enabled == 4 and transaction is not None for enabled, transaction in observed)
+        assert len(bridge.notes) == 1
+        assert bridge.notes[0] is not None
+
+
 class TestInvalidationContract:
     def test_the_bridge_is_told_inside_the_emitting_transaction(self) -> None:
         """Told afterwards, it would be told about a terminal that has changed again."""

tests/test_reserved_lifecycle.py

@@ -211,6 +211,9 @@ def test_reserved_mode_serializes_output_instead_of_proxying_it(self) -> None:
                 assert display._proxy is None
                 assert display._streams
                 assert all(stream.serializer is not None for stream in display._streams)
+                # Commands must reach the qualified outer backend, whose flush enables VT
+                # processing on Windows, rather than writing to its stream or inner writer.
+                assert all(stream.serializer._output is harness.backend for stream in display._streams)
         finally:
             harness.close()