gh-124096: Enable REPL virtual terminal support on Windows (#124119)

To support virtual terminal mode in Windows PYREPL, we need a scanner
to read over the supported escaped VT sequences.

Windows REPL input was using virtual key mode, which does not support
terminal escape sequences. This patch calls `SetConsoleMode` properly
when initializing and send sequences to enable bracketed-paste modes
to support verbatim copy-and-paste.

Signed-off-by: y5c4l3 <y5c4l3@proton.me>
Co-authored-by: Petr Viktorin <encukou@gmail.com>
Co-authored-by: Pablo Galindo Salgado <Pablogsal@gmail.com>
Co-authored-by: Dustin L. Howett <dustin@howett.net>
Co-authored-by: wheeheee <104880306+wheeheee@users.noreply.github.com>

Lib/_pyrepl/base_eventqueue.py

@@ -0,0 +1,108 @@
+#   Copyright 2000-2008 Michael Hudson-Doyle <micahel@gmail.com>
+#                       Armin Rigo
+#
+#                        All Rights Reserved
+#
+#
+# Permission to use, copy, modify, and distribute this software and
+# its documentation for any purpose is hereby granted without fee,
+# provided that the above copyright notice appear in all copies and
+# that both that copyright notice and this permission notice appear in
+# supporting documentation.
+#
+# THE AUTHOR MICHAEL HUDSON DISCLAIMS ALL WARRANTIES WITH REGARD TO
+# THIS SOFTWARE, INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
+# AND FITNESS, IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
+# INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
+# RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
+# CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
+# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+
+"""
+OS-independent base for an event and VT sequence scanner
+
+See unix_eventqueue and windows_eventqueue for subclasses.
+"""
+
+from collections import deque
+
+from . import keymap
+from .console import Event
+from .trace import trace
+
+class BaseEventQueue:
+    def __init__(self, encoding: str, keymap_dict: dict[bytes, str]) -> None:
+        self.compiled_keymap = keymap.compile_keymap(keymap_dict)
+        self.keymap = self.compiled_keymap
+        trace("keymap {k!r}", k=self.keymap)
+        self.encoding = encoding
+        self.events: deque[Event] = deque()
+        self.buf = bytearray()
+
+    def get(self) -> Event | None:
+        """
+        Retrieves the next event from the queue.
+        """
+        if self.events:
+            return self.events.popleft()
+        else:
+            return None
+
+    def empty(self) -> bool:
+        """
+        Checks if the queue is empty.
+        """
+        return not self.events
+
+    def flush_buf(self) -> bytearray:
+        """
+        Flushes the buffer and returns its contents.
+        """
+        old = self.buf
+        self.buf = bytearray()
+        return old
+
+    def insert(self, event: Event) -> None:
+        """
+        Inserts an event into the queue.
+        """
+        trace('added event {event}', event=event)
+        self.events.append(event)
+
+    def push(self, char: int | bytes) -> None:
+        """
+        Processes a character by updating the buffer and handling special key mappings.
+        """
+        ord_char = char if isinstance(char, int) else ord(char)
+        char = bytes(bytearray((ord_char,)))
+        self.buf.append(ord_char)
+        if char in self.keymap:
+            if self.keymap is self.compiled_keymap:
+                # sanity check, buffer is empty when a special key comes
+                assert len(self.buf) == 1
+            k = self.keymap[char]
+            trace('found map {k!r}', k=k)
+            if isinstance(k, dict):
+                self.keymap = k
+            else:
+                self.insert(Event('key', k, self.flush_buf()))
+                self.keymap = self.compiled_keymap
+
+        elif self.buf and self.buf[0] == 27:  # escape
+            # escape sequence not recognized by our keymap: propagate it
+            # outside so that i can be recognized as an M-... key (see also
+            # the docstring in keymap.py
+            trace('unrecognized escape sequence, propagating...')
+            self.keymap = self.compiled_keymap
+            self.insert(Event('key', '\033', bytearray(b'\033')))
+            for _c in self.flush_buf()[1:]:
+                self.push(_c)
+
+        else:
+            try:
+                decoded = bytes(self.buf).decode(self.encoding)
+            except UnicodeError:
+                return
+            else:
+                self.insert(Event('key', decoded, self.flush_buf()))
+            self.keymap = self.compiled_keymap

Lib/_pyrepl/unix_eventqueue.py

@@ -18,12 +18,9 @@
 # CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 
-from collections import deque
-
-from . import keymap
-from .console import Event
 from . import curses
 from .trace import trace
+from .base_eventqueue import BaseEventQueue
 from termios import tcgetattr, VERASE
 import os
 
@@ -70,83 +67,10 @@ def get_terminal_keycodes() -> dict[bytes, str]:
     keycodes.update(CTRL_ARROW_KEYCODES)
     return keycodes
 
-class EventQueue:
+class EventQueue(BaseEventQueue):
     def __init__(self, fd: int, encoding: str) -> None:
-        self.keycodes = get_terminal_keycodes()
+        keycodes = get_terminal_keycodes()
         if os.isatty(fd):
             backspace = tcgetattr(fd)[6][VERASE]
-            self.keycodes[backspace] = "backspace"
+            keycodes[backspace] = "backspace"
-        self.compiled_keymap = keymap.compile_keymap(self.keycodes)
+        BaseEventQueue.__init__(self, encoding, keycodes)
-        self.keymap = self.compiled_keymap
-        trace("keymap {k!r}", k=self.keymap)
-        self.encoding = encoding
-        self.events: deque[Event] = deque()
-        self.buf = bytearray()
-
-    def get(self) -> Event | None:
-        """
-        Retrieves the next event from the queue.
-        """
-        if self.events:
-            return self.events.popleft()
-        else:
-            return None
-
-    def empty(self) -> bool:
-        """
-        Checks if the queue is empty.
-        """
-        return not self.events
-
-    def flush_buf(self) -> bytearray:
-        """
-        Flushes the buffer and returns its contents.
-        """
-        old = self.buf
-        self.buf = bytearray()
-        return old
-
-    def insert(self, event: Event) -> None:
-        """
-        Inserts an event into the queue.
-        """
-        trace('added event {event}', event=event)
-        self.events.append(event)
-
-    def push(self, char: int | bytes) -> None:
-        """
-        Processes a character by updating the buffer and handling special key mappings.
-        """
-        ord_char = char if isinstance(char, int) else ord(char)
-        char = bytes(bytearray((ord_char,)))
-        self.buf.append(ord_char)
-        if char in self.keymap:
-            if self.keymap is self.compiled_keymap:
-                #sanity check, buffer is empty when a special key comes
-                assert len(self.buf) == 1
-            k = self.keymap[char]
-            trace('found map {k!r}', k=k)
-            if isinstance(k, dict):
-                self.keymap = k
-            else:
-                self.insert(Event('key', k, self.flush_buf()))
-                self.keymap = self.compiled_keymap
-
-        elif self.buf and self.buf[0] == 27:  # escape
-            # escape sequence not recognized by our keymap: propagate it
-            # outside so that i can be recognized as an M-... key (see also
-            # the docstring in keymap.py
-            trace('unrecognized escape sequence, propagating...')
-            self.keymap = self.compiled_keymap
-            self.insert(Event('key', '\033', bytearray(b'\033')))
-            for _c in self.flush_buf()[1:]:
-                self.push(_c)
-
-        else:
-            try:
-                decoded = bytes(self.buf).decode(self.encoding)
-            except UnicodeError:
-                return
-            else:
-                self.insert(Event('key', decoded, self.flush_buf()))
-            self.keymap = self.compiled_keymap

Lib/_pyrepl/windows_console.py

@@ -42,6 +42,7 @@ from ctypes import Structure, POINTER, Union
 from .console import Event, Console
 from .trace import trace
 from .utils import wlen
+from .windows_eventqueue import EventQueue
 
 try:
     from ctypes import GetLastError, WinDLL, windll, WinError  # type: ignore[attr-defined]
@@ -94,7 +95,9 @@ VK_MAP: dict[int, str] = {
     0x83: "f20",  # VK_F20
 }
 
-# Console escape codes: https://learn.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences
+# Virtual terminal output sequences
+# Reference: https://learn.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences#output-sequences
+# Check `windows_eventqueue.py` for input sequences
 ERASE_IN_LINE = "\x1b[K"
 MOVE_LEFT = "\x1b[{}D"
 MOVE_RIGHT = "\x1b[{}C"
@@ -110,6 +113,12 @@ CTRL_ACTIVE = 0x04 | 0x08
 class _error(Exception):
     pass
 
+def _supports_vt():
+    try:
+        import nt
+        return nt._supports_virtual_terminal()
+    except (ImportError, AttributeError):
+        return False
 
 class WindowsConsole(Console):
     def __init__(
@@ -121,17 +130,29 @@ class WindowsConsole(Console):
     ):
         super().__init__(f_in, f_out, term, encoding)
 
+        self.__vt_support = _supports_vt()
+
+        if self.__vt_support:
+            trace('console supports virtual terminal')
+
+        # Save original console modes so we can recover on cleanup.
+        original_input_mode = DWORD()
+        GetConsoleMode(InHandle, original_input_mode)
+        trace(f'saved original input mode 0x{original_input_mode.value:x}')
+        self.__original_input_mode = original_input_mode.value
+
         SetConsoleMode(
             OutHandle,
             ENABLE_WRAP_AT_EOL_OUTPUT
             | ENABLE_PROCESSED_OUTPUT
             | ENABLE_VIRTUAL_TERMINAL_PROCESSING,
         )
+
         self.screen: list[str] = []
         self.width = 80
         self.height = 25
         self.__offset = 0
-        self.event_queue: deque[Event] = deque()
+        self.event_queue = EventQueue(encoding)
         try:
             self.out = io._WindowsConsoleIO(self.output_fd, "w")  # type: ignore[attr-defined]
         except ValueError:
@@ -295,6 +316,12 @@ class WindowsConsole(Console):
     def _disable_blinking(self):
         self.__write("\x1b[?12l")
 
+    def _enable_bracketed_paste(self) -> None:
+        self.__write("\x1b[?2004h")
+
+    def _disable_bracketed_paste(self) -> None:
+        self.__write("\x1b[?2004l")
+
     def __write(self, text: str) -> None:
         if "\x1a" in text:
             text = ''.join(["^Z" if x == '\x1a' else x for x in text])
@@ -324,8 +351,15 @@ class WindowsConsole(Console):
         self.__gone_tall = 0
         self.__offset = 0
 
+        if self.__vt_support:
+            SetConsoleMode(InHandle, self.__original_input_mode | ENABLE_VIRTUAL_TERMINAL_INPUT)
+            self._enable_bracketed_paste()
+
     def restore(self) -> None:
-        pass
+        if self.__vt_support:
+            # Recover to original mode before running REPL
+            self._disable_bracketed_paste()
+            SetConsoleMode(InHandle, self.__original_input_mode)
 
     def _move_relative(self, x: int, y: int) -> None:
         """Moves relative to the current posxy"""
@@ -346,7 +380,7 @@ class WindowsConsole(Console):
             raise ValueError(f"Bad cursor position {x}, {y}")
 
         if y < self.__offset or y >= self.__offset + self.height:
-            self.event_queue.insert(0, Event("scroll", ""))
+            self.event_queue.insert(Event("scroll", ""))
         else:
             self._move_relative(x, y)
             self.posxy = x, y
@@ -394,10 +428,8 @@ class WindowsConsole(Console):
         """Return an Event instance.  Returns None if |block| is false
         and there is no event pending, otherwise waits for the
         completion of an event."""
-        if self.event_queue:
-            return self.event_queue.pop()
 
-        while True:
+        while self.event_queue.empty():
             rec = self._read_input(block)
             if rec is None:
                 return None
@@ -428,20 +460,25 @@ class WindowsConsole(Console):
                         key = f"ctrl {key}"
                     elif key_event.dwControlKeyState & ALT_ACTIVE:
                         # queue the key, return the meta command
-                        self.event_queue.insert(0, Event(evt="key", data=key, raw=key))
+                        self.event_queue.insert(Event(evt="key", data=key, raw=key))
                         return Event(evt="key", data="\033")  # keymap.py uses this for meta
                     return Event(evt="key", data=key, raw=key)
                 if block:
                     continue
 
                 return None
+            elif self.__vt_support:
+                # If virtual terminal is enabled, scanning VT sequences
+                self.event_queue.push(rec.Event.KeyEvent.uChar.UnicodeChar)
+                continue
 
             if key_event.dwControlKeyState & ALT_ACTIVE:
                 # queue the key, return the meta command
-                self.event_queue.insert(0, Event(evt="key", data=key, raw=raw_key))
+                self.event_queue.insert(Event(evt="key", data=key, raw=raw_key))
                 return Event(evt="key", data="\033")  # keymap.py uses this for meta
 
             return Event(evt="key", data=key, raw=raw_key)
+        return self.event_queue.get()
 
     def push_char(self, char: int | bytes) -> None:
         """
@@ -563,6 +600,13 @@ MENU_EVENT = 0x08
 MOUSE_EVENT = 0x02
 WINDOW_BUFFER_SIZE_EVENT = 0x04
 
+ENABLE_PROCESSED_INPUT = 0x0001
+ENABLE_LINE_INPUT = 0x0002
+ENABLE_ECHO_INPUT = 0x0004
+ENABLE_MOUSE_INPUT = 0x0010
+ENABLE_INSERT_MODE = 0x0020
+ENABLE_VIRTUAL_TERMINAL_INPUT = 0x0200
+
 ENABLE_PROCESSED_OUTPUT = 0x01
 ENABLE_WRAP_AT_EOL_OUTPUT = 0x02
 ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x04
@@ -594,6 +638,10 @@ if sys.platform == "win32":
     ]
     ScrollConsoleScreenBuffer.restype = BOOL
 
+    GetConsoleMode = _KERNEL32.GetConsoleMode
+    GetConsoleMode.argtypes = [HANDLE, POINTER(DWORD)]
+    GetConsoleMode.restype = BOOL
+
     SetConsoleMode = _KERNEL32.SetConsoleMode
     SetConsoleMode.argtypes = [HANDLE, DWORD]
     SetConsoleMode.restype = BOOL
@@ -620,6 +668,7 @@ else:
     GetStdHandle = _win_only
     GetConsoleScreenBufferInfo = _win_only
     ScrollConsoleScreenBuffer = _win_only
+    GetConsoleMode = _win_only
     SetConsoleMode = _win_only
     ReadConsoleInput = _win_only
     GetNumberOfConsoleInputEvents = _win_only

Lib/_pyrepl/windows_eventqueue.py

@@ -0,0 +1,42 @@
+"""
+Windows event and VT sequence scanner
+"""
+
+from .base_eventqueue import BaseEventQueue
+
+
+# Reference: https://learn.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences#input-sequences
+VT_MAP: dict[bytes, str] = {
+    b'\x1b[A': 'up',
+    b'\x1b[B': 'down',
+    b'\x1b[C': 'right',
+    b'\x1b[D': 'left',
+    b'\x1b[1;5D': 'ctrl left',
+    b'\x1b[1;5C': 'ctrl right',
+
+    b'\x1b[H': 'home',
+    b'\x1b[F': 'end',
+
+    b'\x7f': 'backspace',
+    b'\x1b[2~': 'insert',
+    b'\x1b[3~': 'delete',
+    b'\x1b[5~': 'page up',
+    b'\x1b[6~': 'page down',
+
+    b'\x1bOP':   'f1',
+    b'\x1bOQ':   'f2',
+    b'\x1bOR':   'f3',
+    b'\x1bOS':   'f4',
+    b'\x1b[15~': 'f5',
+    b'\x1b[17~': 'f6',
+    b'\x1b[18~': 'f7',
+    b'\x1b[19~': 'f8',
+    b'\x1b[20~': 'f9',
+    b'\x1b[21~': 'f10',
+    b'\x1b[23~': 'f11',
+    b'\x1b[24~': 'f12',
+}
+
+class EventQueue(BaseEventQueue):
+    def __init__(self, encoding: str) -> None:
+        BaseEventQueue.__init__(self, encoding, VT_MAP)

Lib/test/test_pyrepl/test_eventqueue.py

@@ -2,70 +2,77 @@ import tempfile
 import unittest
 import sys
 from unittest.mock import patch
+from test import support
 
 try:
     from _pyrepl.console import Event
-    from _pyrepl.unix_eventqueue import EventQueue
+    from _pyrepl import base_eventqueue
 except ImportError:
     pass
 
-@unittest.skipIf(sys.platform == "win32", "No Unix event queue on Windows")
+try:
-@patch("_pyrepl.curses.tigetstr", lambda x: b"")
+    from _pyrepl import unix_eventqueue
-class TestUnixEventQueue(unittest.TestCase):
+except ImportError:
-    def setUp(self):
+    pass
-        self.file = tempfile.TemporaryFile()
 
-    def tearDown(self) -> None:
+try:
-        self.file.close()
+    from _pyrepl import windows_eventqueue
+except ImportError:
+    pass
+
+class EventQueueTestBase:
+    """OS-independent mixin"""
+    def make_eventqueue(self) -> base_eventqueue.BaseEventQueue:
+        raise NotImplementedError()
 
     def test_get(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         event = Event("key", "a", b"a")
         eq.insert(event)
         self.assertEqual(eq.get(), event)
 
     def test_empty(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         self.assertTrue(eq.empty())
         eq.insert(Event("key", "a", b"a"))
         self.assertFalse(eq.empty())
 
     def test_flush_buf(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.buf.extend(b"test")
         self.assertEqual(eq.flush_buf(), b"test")
         self.assertEqual(eq.buf, bytearray())
 
     def test_insert(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         event = Event("key", "a", b"a")
         eq.insert(event)
         self.assertEqual(eq.events[0], event)
 
-    @patch("_pyrepl.unix_eventqueue.keymap")
+    @patch("_pyrepl.base_eventqueue.keymap")
     def test_push_with_key_in_keymap(self, mock_keymap):
         mock_keymap.compile_keymap.return_value = {"a": "b"}
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {b"a": "b"}
         eq.push("a")
         mock_keymap.compile_keymap.assert_called()
         self.assertEqual(eq.events[0].evt, "key")
         self.assertEqual(eq.events[0].data, "b")
 
-    @patch("_pyrepl.unix_eventqueue.keymap")
+    @patch("_pyrepl.base_eventqueue.keymap")
     def test_push_without_key_in_keymap(self, mock_keymap):
         mock_keymap.compile_keymap.return_value = {"a": "b"}
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {b"c": "d"}
         eq.push("a")
         mock_keymap.compile_keymap.assert_called()
         self.assertEqual(eq.events[0].evt, "key")
         self.assertEqual(eq.events[0].data, "a")
 
-    @patch("_pyrepl.unix_eventqueue.keymap")
+    @patch("_pyrepl.base_eventqueue.keymap")
     def test_push_with_keymap_in_keymap(self, mock_keymap):
         mock_keymap.compile_keymap.return_value = {"a": "b"}
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {b"a": {b"b": "c"}}
         eq.push("a")
         mock_keymap.compile_keymap.assert_called()
@@ -77,10 +84,10 @@ class TestUnixEventQueue(unittest.TestCase):
         self.assertEqual(eq.events[1].evt, "key")
         self.assertEqual(eq.events[1].data, "d")
 
-    @patch("_pyrepl.unix_eventqueue.keymap")
+    @patch("_pyrepl.base_eventqueue.keymap")
     def test_push_with_keymap_in_keymap_and_escape(self, mock_keymap):
         mock_keymap.compile_keymap.return_value = {"a": "b"}
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {b"a": {b"b": "c"}}
         eq.push("a")
         mock_keymap.compile_keymap.assert_called()
@@ -94,7 +101,7 @@ class TestUnixEventQueue(unittest.TestCase):
         self.assertEqual(eq.events[1].data, "b")
 
     def test_push_special_key(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {}
         eq.push("\x1b")
         eq.push("[")
@@ -103,7 +110,7 @@ class TestUnixEventQueue(unittest.TestCase):
         self.assertEqual(eq.events[0].data, "\x1b")
 
     def test_push_unrecognized_escape_sequence(self):
-        eq = EventQueue(self.file.fileno(), "utf-8")
+        eq = self.make_eventqueue()
         eq.keymap = {}
         eq.push("\x1b")
         eq.push("[")
@@ -115,3 +122,22 @@ class TestUnixEventQueue(unittest.TestCase):
         self.assertEqual(eq.events[1].data, "[")
         self.assertEqual(eq.events[2].evt, "key")
         self.assertEqual(eq.events[2].data, "Z")
+
+
+@unittest.skipIf(support.MS_WINDOWS, "No Unix event queue on Windows")
+class TestUnixEventQueue(EventQueueTestBase, unittest.TestCase):
+    def setUp(self):
+        self.enterContext(patch("_pyrepl.curses.tigetstr", lambda x: b""))
+        self.file = tempfile.TemporaryFile()
+
+    def tearDown(self) -> None:
+        self.file.close()
+
+    def make_eventqueue(self) -> base_eventqueue.BaseEventQueue:
+        return unix_eventqueue.EventQueue(self.file.fileno(), "utf-8")
+
+
+@unittest.skipUnless(support.MS_WINDOWS, "No Windows event queue on Unix")
+class TestWindowsEventQueue(EventQueueTestBase, unittest.TestCase):
+    def make_eventqueue(self) -> base_eventqueue.BaseEventQueue:
+        return windows_eventqueue.EventQueue("utf-8")

Misc/NEWS.d/next/Library/2024-09-16-17-03-52.gh-issue-124096.znin0O.rst

@@ -0,0 +1,3 @@
+Turn on virtual terminal mode and enable bracketed paste in REPL on Windows
+console. (If the terminal does not support bracketed paste, enabling it
+does nothing.)