gh-154730: Add timeout in PyREPL to distinguish between bare Esc and multibyte escape sequence

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()
+        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.

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))

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)

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

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)

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.