Switch to pyrepl

.github/workflows/ci.yml

@@ -301,6 +301,7 @@ jobs:
         run: ./ci.sh
         env:
           NO_TEST_REQUIREMENTS: '${{ matrix.no_test_requirements }}'
+          TERM: xterm
       - if: >-
           always()
           && matrix.check_formatting != '1'
@@ -346,6 +347,8 @@ jobs:
           check-latest: true
       - name: Run tests
         run: ./ci.sh
+        env:
+          TERM: xterm
       - if: always()
         uses: codecov/codecov-action@v5
         with:

newsfragments/3007.bugfix.rst

@@ -0,0 +1,6 @@
+Switch ``python -m trio`` to use ``pyrepl``. This means that keyboard
+interrupts should be processed more consistently, as well as matching
+the builtin Python REPL more. One downside is that now using the trio
+REPL requires PyPy, CPython 3.13+, or installing the ``pyrepl`` package
+from PyPI. We recommend the first two options, as our support for the
+version on PyPI is worse.

src/trio/_repl.py

@@ -1,48 +1,51 @@
 from __future__ import annotations
 
 import ast
-import contextlib
 import inspect
 import sys
-import warnings
 from code import InteractiveConsole
-from types import CodeType, FrameType, FunctionType
-from typing import TYPE_CHECKING
+from signal import SIGINT, raise_signal
+from types import CodeType, FunctionType
+from typing import cast
 
 import outcome
 
 import trio
-import trio.lowlevel
 from trio._util import final
 
-if TYPE_CHECKING:
-    from collections.abc import Callable
-
-
-class SuppressDecorator(contextlib.ContextDecorator, contextlib.suppress):
-    pass
-
-
-@SuppressDecorator(KeyboardInterrupt)
-@trio.lowlevel.disable_ki_protection
-def terminal_newline() -> None:  # TODO: test this line
-    import fcntl
-    import termios
-
-    # Fake up a newline char as if user had typed it at the terminal
+try:
+    import pyrepl
+    from pyrepl import commands, reader as r, readline
+except ImportError:
     try:
-        fcntl.ioctl(sys.stdin, termios.TIOCSTI, b"\n")  # type: ignore[attr-defined, unused-ignore]
-    except OSError as e:
-        print(f"\nPress enter! Newline injection failed: {e}", end="", flush=True)
+        import _pyrepl as pyrepl
+        from _pyrepl import commands, reader as r, readline
+    except ImportError:
+        print(
+            "Trio's REPL requires CPython 3.13+, PyPy, or installing pyrepl from PyPI."
+        )
+        exit(1)
+
+# there are differences between the CPython pyrepl and PyPI pyrepl
+try:
+    # The following expression fails on PyPy, even though you can
+    # `import pyrepl`. This is important because PyPy simply vendors
+    # CPython pyrepl: https://github.com/pypy/pypy/issues/4990
+    pyrepl.__version__  # noqa: B018
+except AttributeError:
+    CPYTHON_VENDOR = True
+else:
+    CPYTHON_VENDOR = False
 
 
 @final
 class TrioInteractiveConsole(InteractiveConsole):
-    def __init__(self, repl_locals: dict[str, object] | None = None) -> None:
+    def __init__(
+        self,
+        repl_locals: dict[str, object] | None = None,
+    ) -> None:
         super().__init__(locals=repl_locals)
-        self.token: trio.lowlevel.TrioToken | None = None
         self.compile.compiler.flags |= ast.PyCF_ALLOW_TOP_LEVEL_AWAIT
-        self.interrupted = False
 
     def runcode(self, code: CodeType) -> None:
         func = FunctionType(code, self.locals)
@@ -71,79 +74,71 @@ def runcode(self, code: CodeType) -> None:
                 # We always use sys.excepthook, unlike other implementations.
                 # This means that overriding self.write also does nothing to tbs.
                 sys.excepthook(sys.last_type, sys.last_value, sys.last_traceback)
+
         # clear any residual KI
         trio.from_thread.run(trio.lowlevel.checkpoint_if_cancelled)
         # trio.from_thread.check_cancelled() has too long of a memory
 
-    if sys.platform == "win32":  # TODO: test this line
 
-        def raw_input(self, prompt: str = "") -> str:
-            try:
-                return input(prompt)
-            except EOFError:
-                # check if trio has a pending KI
-                trio.from_thread.run(trio.lowlevel.checkpoint_if_cancelled)
-                raise
-
-    else:
-
-        def raw_input(self, prompt: str = "") -> str:
-            from signal import SIGINT, signal
-
-            assert not self.interrupted
-
-            def install_handler() -> (
-                Callable[[int, FrameType | None], None] | int | None
-            ):
-                def handler(
-                    sig: int, frame: FrameType | None
-                ) -> None:  # TODO: test this line
-                    self.interrupted = True
-                    token.run_sync_soon(terminal_newline, idempotent=True)
-
-                token = trio.lowlevel.current_trio_token()
-
-                return signal(SIGINT, handler)
+async def repl_input(reader: r.Reader | None, prompt: str) -> str:  # type: ignore[no-any-unimported]
+    assert reader is not None
+    reader.ps1 = prompt
+    reader.prepare()
+    try:
+        reader.refresh()
+        while not reader.finished:
+            if not reader.handle1(block=False):
+                if sys.platform == "win32":  # TODO: test this line
+                    await trio.lowlevel.wait_readable(pyrepl.windows_console.InHandle)
+                else:
+                    await trio.lowlevel.wait_readable(reader.console.input_fd)
+
+        if CPYTHON_VENDOR:
+            return reader.get_unicode()  # type: ignore[no-any-return]
+        else:
+            return reader.get_str()  # type: ignore[no-any-return]
+    finally:
+        reader.restore()
 
-            prev_handler = trio.from_thread.run_sync(install_handler)
-            try:
-                return input(prompt)
-            finally:
-                trio.from_thread.run_sync(signal, SIGINT, prev_handler)
-                if self.interrupted:  # TODO: test this line
-                    raise KeyboardInterrupt
-
-        def write(self, output: str) -> None:
-            if self.interrupted:  # TODO: test this line
-                assert output == "\nKeyboardInterrupt\n"
-                sys.stderr.write(output[1:])
-                self.interrupted = False
-            else:
-                sys.stderr.write(output)
 
+async def run_repl(console: TrioInteractiveConsole, reader: r.Reader | None) -> None:  # type: ignore[no-any-unimported]
+    # mostly copy-pasted from code.InteractiveConsole.interact
+    try:
+        sys.ps1  # noqa: B018
+    except AttributeError:
+        sys.ps1 = ">>> "
+    try:
+        sys.ps2  # noqa: B018
+    except AttributeError:
+        sys.ps2 = "... "
 
-async def run_repl(console: TrioInteractiveConsole) -> None:
     banner = (
         f"trio REPL {sys.version} on {sys.platform}\n"
         f'Use "await" directly instead of "trio.run()".\n'
         f'Type "help", "copyright", "credits" or "license" '
         f"for more information.\n"
-        f'{getattr(sys, "ps1", ">>> ")}import trio'
+        f'{getattr(sys, "ps1", ">>> ")}import trio\n'
     )
-    try:
-        await trio.to_thread.run_sync(console.interact, banner)
-    finally:
-        warnings.filterwarnings(
-            "ignore",
-            message=r"^coroutine .* was never awaited$",
-            category=RuntimeWarning,
-        )
+    console.write(banner)
+    more = 0
 
+    while True:
+        try:
+            prompt = cast("str", sys.ps2 if more else sys.ps1)
+            try:
+                line = await repl_input(reader, prompt)
+            except EOFError:
+                console.write("\n")
+                break
+            else:
+                more = await trio.to_thread.run_sync(console.push, line)
+        except KeyboardInterrupt:  # TODO: test this line
+            console.write("\nKeyboardInterrupt\n")
+            console.resetbuffer()
+            more = 0
 
-def main(original_locals: dict[str, object]) -> None:
-    with contextlib.suppress(ImportError):
-        import readline  # noqa: F401
 
+def main(original_locals: dict[str, object]) -> None:
     repl_locals: dict[str, object] = {"trio": trio}
     for key in {
         "__name__",
@@ -155,5 +150,19 @@ def main(original_locals: dict[str, object]) -> None:
     }:
         repl_locals[key] = original_locals[key]
 
+    # This call also registers all necessary signal handlers.
+    # Otherwise, we would not be able to run `readline` in a child
+    # thread.
+    reader = readline._get_reader()
+
+    if not CPYTHON_VENDOR:
+        # The default `interrupt` command finishes the console, which
+        # adds an extra newline. Unforgivable!
+        class interrupt(commands.FinishCommand):  # type: ignore[misc,no-any-unimported]
+            def do(self) -> None:  # TODO: test this line
+                raise_signal(SIGINT)
+
+        reader.commands["interrupt"] = interrupt
+
     console = TrioInteractiveConsole(repl_locals)
-    trio.run(run_repl, console)
+    trio.run(run_repl, console, reader)