diff --git a/Lib/_pyrepl/base_eventqueue.py b/Lib/_pyrepl/base_eventqueue.py index f520ffd7d8ad11d..a3353cb07e4e906 100644 --- a/Lib/_pyrepl/base_eventqueue.py +++ b/Lib/_pyrepl/base_eventqueue.py @@ -24,20 +24,26 @@ See unix_eventqueue and windows_eventqueue for subclasses. """ +import os from collections import deque from . import keymap from .console import Event from .trace import trace +ESC_TIMEOUT_DEFAULT = 0.1 + class BaseEventQueue: - def __init__(self, encoding: str, keymap_dict: dict[bytes, str]) -> None: + def __init__(self, encoding: str, keymap_dict: dict[bytes, str], + esc_timeout: float | None = None) -> 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() + default = float(os.environ.get("PYREPL_ESC_TIMEOUT", ESC_TIMEOUT_DEFAULT)) + self.esc_timeout = esc_timeout if esc_timeout is not None else default def get(self) -> Event | None: """ @@ -54,6 +60,39 @@ def empty(self) -> bool: """ return not self.events + def pending(self) -> bool: + """ + True when we have received the start of an escape/key sequence but + are still waiting for further bytes to disambiguate it. + + This is the case right after a byte (such as ESC) that our keymap + knows only as a prefix of a longer sequence has been pushed, but + before the rest of that sequence arrives. Consoles use this to + decide whether the next read should block indefinitely or only for + the escape timeout. + """ + return self.keymap is not self.compiled_keymap + + def flush(self) -> None: + """ + Finalize any pending, incomplete input as discrete events. + """ + if self.keymap is self.compiled_keymap: + # Nothing pending: either idle, or a complete key was already + # emitted by ``push``. + return + buf = self.buf.take_bytes() # type: ignore[attr-defined] + self.keymap = self.compiled_keymap + if buf and buf[0] == 27: # escape + self.insert(Event('key', '\033', b'\033')) + for _c in buf[1:]: + self.push(_c) + else: + # Defensive: a pending prefix that does not start with ESC + # (e.g. a partial multi-byte sequence). Emit it as text. + data = bytes(buf).decode(self.encoding, 'replace') + self.insert(Event('key', data, buf)) + def insert(self, event: Event) -> None: """ Inserts an event into the queue. diff --git a/Lib/_pyrepl/unix_console.py b/Lib/_pyrepl/unix_console.py index 8edd1c36a7232d7..23cfe27718d6f44 100644 --- a/Lib/_pyrepl/unix_console.py +++ b/Lib/_pyrepl/unix_console.py @@ -558,6 +558,12 @@ def get_event(self, block: bool = True) -> Event | None: return None while self.event_queue.empty(): + if self.event_queue.pending(): + if not self.wait(timeout=self.event_queue.esc_timeout): + # Timed out waiting for the rest of a sequence: flush the + # pending prefix as a complete key (e.g. a lone ESC). + self.event_queue.flush() + continue while True: try: self.push_char(self.__read(1)) diff --git a/Lib/_pyrepl/unix_eventqueue.py b/Lib/_pyrepl/unix_eventqueue.py index 2a9cca59e7477f7..cd0364746680c84 100644 --- a/Lib/_pyrepl/unix_eventqueue.py +++ b/Lib/_pyrepl/unix_eventqueue.py @@ -69,9 +69,10 @@ def get_terminal_keycodes(ti: TermInfo) -> dict[bytes, str]: class EventQueue(BaseEventQueue): - def __init__(self, fd: int, encoding: str, ti: TermInfo) -> None: + def __init__(self, fd: int, encoding: str, ti: TermInfo, + esc_timeout: float | None = None) -> None: keycodes = get_terminal_keycodes(ti) if os.isatty(fd): backspace = tcgetattr(fd)[6][VERASE] keycodes[backspace] = "backspace" - BaseEventQueue.__init__(self, encoding, keycodes) + BaseEventQueue.__init__(self, encoding, keycodes, esc_timeout) diff --git a/Lib/_pyrepl/windows_console.py b/Lib/_pyrepl/windows_console.py index 3768a22ad16f7bb..2747b736bcd9413 100644 --- a/Lib/_pyrepl/windows_console.py +++ b/Lib/_pyrepl/windows_console.py @@ -572,6 +572,12 @@ def get_event(self, block: bool = True) -> Event | None: return None while self.event_queue.empty(): + if self.event_queue.pending(): + # Timed out waiting for the rest of a sequence: flush the + # pending prefix as a complete key (e.g. a lone ESC). + if not self.wait_for_event(self.event_queue.esc_timeout * 1000): + self.event_queue.flush() + continue rec = self._read_input() if rec is None: return None diff --git a/Lib/_pyrepl/windows_eventqueue.py b/Lib/_pyrepl/windows_eventqueue.py index d99722f9a16a93a..19496c2bb02280b 100644 --- a/Lib/_pyrepl/windows_eventqueue.py +++ b/Lib/_pyrepl/windows_eventqueue.py @@ -38,5 +38,5 @@ } class EventQueue(BaseEventQueue): - def __init__(self, encoding: str) -> None: - BaseEventQueue.__init__(self, encoding, VT_MAP) + def __init__(self, encoding: str, esc_timeout: float | None = None) -> None: + BaseEventQueue.__init__(self, encoding, VT_MAP, esc_timeout) diff --git a/Lib/test/test_pyrepl/test_eventqueue.py b/Lib/test/test_pyrepl/test_eventqueue.py index 56ab43211847a49..dbabb83d2aa06b2 100644 --- a/Lib/test/test_pyrepl/test_eventqueue.py +++ b/Lib/test/test_pyrepl/test_eventqueue.py @@ -1,3 +1,4 @@ +import os import tempfile import unittest from unittest.mock import patch @@ -167,6 +168,37 @@ def _push(keys): self.assertEqual(eq.get(), _event("key", "b")) self.assertEqual(eq.get(), _event("key", "a")) + def test_pending(self): + eq = base_eventqueue.BaseEventQueue("utf-8", {}) + eq.compiled_keymap = eq.keymap = {b"\x1b": {b"[": "down"}} + self.assertFalse(eq.pending()) + eq.push(b"\x1b") + self.assertTrue(eq.pending()) + eq.push(b"[") + self.assertFalse(eq.pending()) + self.assertEqual(eq.get().data, "down") + + def test_flush(self): + eq = base_eventqueue.BaseEventQueue("utf-8", {}) + eq.compiled_keymap = eq.keymap = {b"\x1b": {b"[": "down"}} + eq.push(b"\x1b") + self.assertTrue(eq.pending()) + eq.flush() + self.assertFalse(eq.pending()) + event = eq.get() + self.assertIsNotNone(event) + self.assertEqual(event.evt, "key") + self.assertEqual(event.data, "\x1b") + self.assertEqual(event.raw, b"\x1b") + + def test_esc_timeout_override(self): + eq = base_eventqueue.BaseEventQueue("utf-8", {}, esc_timeout=0.3) + self.assertAlmostEqual(eq.esc_timeout, 0.3) + + with patch.dict(os.environ, {"PYREPL_ESC_TIMEOUT": "0.25"}): + eq = base_eventqueue.BaseEventQueue("utf-8", {}) + self.assertAlmostEqual(eq.esc_timeout, 0.25) + class EmptyTermInfo(terminfo.TermInfo): def get(self, cap: str) -> bytes: diff --git a/Lib/test/test_pyrepl/test_unix_console.py b/Lib/test/test_pyrepl/test_unix_console.py index 2fc8398923cbf38..49416d86d919c30 100644 --- a/Lib/test/test_pyrepl/test_unix_console.py +++ b/Lib/test/test_pyrepl/test_unix_console.py @@ -399,6 +399,20 @@ def test_restore_in_thread(self, _os_write): thread.start() thread.join() # this should not raise + def test_escape_timeout(self, _os_write): + console = UnixConsole(term="xterm") + console.prepare() + console.event_queue.push(b"\x1b") + self.assertTrue(console.event_queue.pending()) + # Simulate the read timing out before any follow-up bytes arrive. + console.wait = Mock(return_value=False) + event = console.get_event() + self.assertIsNotNone(event) + self.assertEqual(event.evt, "key") + self.assertEqual(event.data, "\x1b") + self.assertEqual(event.raw, b"\x1b") + console.restore() + @unittest.skipIf(sys.platform == "win32", "No Unix console on Windows") class TestUnixConsoleEIOHandling(TestCase): diff --git a/Lib/test/test_pyrepl/test_windows_console.py b/Lib/test/test_pyrepl/test_windows_console.py index 32c4255aa6b1904..ea8211f20ec67ab 100644 --- a/Lib/test/test_pyrepl/test_windows_console.py +++ b/Lib/test/test_pyrepl/test_windows_console.py @@ -648,6 +648,19 @@ def test_wait_not_empty(self): self.assertTrue(console.wait(0.0)) self.assertEqual(console.wait_for_event.call_count, 0) + def test_escape_timeout(self): + console = WindowsConsole() + # Simulate the read timing out before any follow-up bytes arrive. + console.wait_for_event = MagicMock(return_value=False) + console.event_queue.push(b"\x1b") + self.assertTrue(console.event_queue.pending()) + event = console.get_event() + console.restore() + self.assertIsNotNone(event) + self.assertEqual(event.evt, "key") + self.assertEqual(event.data, "\x1b") + self.assertEqual(event.raw, b"\x1b") + if __name__ == "__main__": unittest.main() diff --git a/Misc/NEWS.d/next/Library/2026-07-26-19-43-35.gh-issue-154730.m5zUk_.rst b/Misc/NEWS.d/next/Library/2026-07-26-19-43-35.gh-issue-154730.m5zUk_.rst new file mode 100644 index 000000000000000..75780ef8879dfdc --- /dev/null +++ b/Misc/NEWS.d/next/Library/2026-07-26-19-43-35.gh-issue-154730.m5zUk_.rst @@ -0,0 +1,2 @@ +Add timeout in :mod:`!_pyrepl` to distinguish between bare ``Esc`` and +multibyte escape sequence.