gh-152907: Restore cooked output flags around the input hook in the new REPL (#153389)

* gh-152907: Restore cooked output flags around the input hook in the new REPL

pyrepl clears OPOST for its own cursor rendering but calls PyOS_InputHook
from inside the raw-mode read loop, so output written by an input hook
(GUI toolkit event loops, and any warning/traceback/print they emit) is
emitted with bare '\n' and no '\r'.  Restore the terminal's saved output
flags around the hook call and re-enter raw mode afterwards; only oflag is
toggled so ECHO/ICANON stay off at the prompt.

* Skip the input-hook test on platforms without pty devices

The Emscripten buildbot has the pty module but no pty devices, so
pty.openpty() raises OSError("out of pty devices").  Guard the test
class the same way Lib/test/test_pty.py does.

* Propagate the input hook's return value and drop the sleep from the test

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Trim comments

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Observe the hook's output synchronously instead of via the reader thread

The reader thread no longer feeds any assertion: the hook drains the pty
master itself, so the check is an exact comparison. A drainer is still
needed for restore(), which writes before switching modes.

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

Author: harjothkhara

Lib/_pyrepl/unix_console.py

@@ -486,6 +486,7 @@ def prepare(self) -> None:
         raw.cc[termios.VMIN] = b"\x01"
         raw.cc[termios.VTIME] = b"\x00"
         self.__input_fd_set(raw)
+        self.__rawtermstate = raw
 
         # Apple Terminal will re-wrap lines for us unless we preempt the
         # damage.
@@ -726,7 +727,19 @@ def input_hook(self):
         # avoid inline imports here so the repl doesn't get flooded
         # with import logging from -X importtime=2
         if posix is not None and posix._is_inputhook_installed():
-            return posix._inputhook
+            return self.__run_input_hook
+
+    def __run_input_hook(self):
+        # gh-152907: input hooks expect cooked output, but pyrepl runs with
+        # OPOST disabled.  Restore the saved output flags around the hook
+        # (only oflag; input must stay raw at the prompt).
+        cooked = self.__rawtermstate.copy()
+        cooked.oflag = self.__svtermstate.oflag
+        self.__input_fd_set(cooked)
+        try:
+            return posix._inputhook()
+        finally:
+            self.__input_fd_set(self.__rawtermstate)
 
     def __enable_bracketed_paste(self) -> None:
         os.write(self.output_fd, b"\x1b[?2004h")

Lib/test/test_pyrepl/test_unix_console.py

@@ -1,13 +1,15 @@
 import errno
 import itertools
 import os
+import select
 import signal
 import sys
 import threading
 import unittest
 from functools import partial
 from _colorize import ANSIColors
 from test.support import force_color, os_helper, force_not_colorized_test_class
+from test.support import is_android, is_apple_mobile, is_wasm32
 from test.support import threading_helper
 
 from unittest import TestCase
@@ -424,3 +426,93 @@ def test_eio_error_handling_in_restore(self, mock_tcgetattr, mock_tcsetattr):
 
         # EIO error should be handled gracefully in restore()
         console.restore()
+
+
+try:
+    import pty
+    import termios as _termios
+except ImportError:
+    pty = None
+
+
+@unittest.skipIf(sys.platform == "win32", "No Unix console on Windows")
+@unittest.skipUnless(pty, "requires pty")
+@unittest.skipIf(is_android or is_apple_mobile or is_wasm32,
+                 "pty is not available on this platform")
+class TestUnixConsoleInputHook(TestCase):
+    # gh-152907: the console must restore cooked output (OPOST) around
+    # input-hook calls, then re-enter raw mode.
+
+    def test_input_hook_output_is_cooked(self):
+        master_fd, slave_fd = pty.openpty()
+        self.addCleanup(os.close, master_fd)
+
+        # tcsetattr(TCSADRAIN) blocks on some platforms (e.g. macOS) while the
+        # master still holds unread output, so empty it before each mode switch.
+        def drain():
+            out = b""
+            while select.select([master_fd], [], [], 0)[0]:
+                try:
+                    data = os.read(master_fd, 4096)
+                except OSError:
+                    break
+                if not data:
+                    break
+                out += data
+            return out
+
+        # Start from a cooked terminal so there are saved flags to restore.
+        attr = _termios.tcgetattr(slave_fd)
+        attr[1] |= _termios.OPOST | _termios.ONLCR
+        _termios.tcsetattr(slave_fd, _termios.TCSANOW, attr)
+
+        console = UnixConsole(slave_fd, slave_fd, term="xterm")
+        console.prepare()
+        try:
+            drain()  # discard prepare()'s own setup sequences
+            # pyrepl's own rendering runs with OPOST cleared.
+            self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)
+
+            observed = {}
+
+            def fake_hook():
+                observed["oflag"] = _termios.tcgetattr(slave_fd)[1]
+                os.write(slave_fd, b"line1\nline2\n")
+                observed["output"] = drain()
+                return 0
+
+            with patch("_pyrepl.unix_console.posix") as mock_posix:
+                mock_posix._is_inputhook_installed.return_value = True
+                mock_posix._inputhook.side_effect = fake_hook
+                hook = console.input_hook
+                self.assertIsNotNone(hook)
+                self.assertEqual(hook(), 0)
+
+            # The hook ran with cooked output (OPOST on)...
+            self.assertTrue(observed["oflag"] & _termios.OPOST)
+            # ...and raw mode was restored afterwards.
+            self.assertFalse(_termios.tcgetattr(slave_fd)[1] & _termios.OPOST)
+            # The tty translated the hook's bare '\n' into '\r\n'.
+            self.assertEqual(observed["output"], b"line1\r\nline2\r\n")
+        finally:
+            # restore() writes and only then switches modes, so there is no
+            # point left to drain from here; keep the master empty elsewhere.
+            stop = threading.Event()
+
+            def pump():
+                while not stop.is_set():
+                    if select.select([master_fd], [], [], 0.05)[0]:
+                        try:
+                            if not os.read(master_fd, 4096):
+                                break
+                        except OSError:
+                            break
+
+            pump_thread = threading.Thread(target=pump)
+            pump_thread.start()
+            try:
+                console.restore()
+            finally:
+                stop.set()
+                pump_thread.join()
+                os.close(slave_fd)

Misc/NEWS.d/next/Library/2026-07-08-23-20-00.gh-issue-152907.oyPV9Y.rst

@@ -0,0 +1,4 @@
+Restore cooked-mode terminal output flags around :c:data:`PyOS_InputHook`
+callbacks in the new :term:`REPL` (:mod:`!_pyrepl`), so that output written
+by an input hook (for example a GUI toolkit event loop) is no longer emitted
+with ``OPOST`` disabled and keeps its carriage returns.