Fix inputhook file descriptor errors

src/prompt_toolkit/application/application.py

@@ -973,10 +973,13 @@ def _called_from_ipython() -> bool:
             # In Python 3.12, we can use asyncio.run(loop_factory=...)
             # For now, use `run_until_complete()`.
             loop = new_eventloop_with_inputhook(inputhook)
-            result = loop.run_until_complete(coro)
-            loop.run_until_complete(loop.shutdown_asyncgens())
-            loop.close()
-            return result
+            try:
+                return loop.run_until_complete(coro)
+            finally:
+                try:
+                    loop.run_until_complete(loop.shutdown_asyncgens())
+                finally:
+                    loop.close()
 
         elif _called_from_ipython():
             # workaround to make input hooks work for IPython until

src/prompt_toolkit/eventloop/inputhook.py

@@ -27,9 +27,7 @@
 
 import asyncio
 import os
-import select
 import selectors
-import sys
 import threading
 from asyncio import AbstractEventLoop, get_running_loop
 from collections.abc import Callable, Mapping
@@ -151,38 +149,26 @@ def input_is_ready() -> bool:
         # own loop, or by checking the `input_is_ready` function regularly.
         self.inputhook(InputHookContext(self._r, input_is_ready))
 
-        # Flush the read end of the pipe.
-        try:
-            # Before calling 'os.read', call select.select. This is required
-            # when the gevent monkey patch has been applied. 'os.read' is never
-            # monkey patched and won't be cooperative, so that would block all
-            # other select() calls otherwise.
-            # See: http://www.gevent.org/gevent.os.html
-
-            # Note: On Windows, this is apparently not an issue.
-            #       However, if we would ever want to add a select call, it
-            #       should use `windll.kernel32.WaitForMultipleObjects`,
-            #       because `select.select` can't wait for a pipe on Windows.
-            if sys.platform != "win32":
-                select.select([self._r], [], [], None)
+        # Wait for the real selector to be done before reading from the pipe.
+        # The selector thread writes to the pipe before setting `ready`.
+        th.join()
 
+        try:
             os.read(self._r, 1024)
         except OSError:
             # This happens when the window resizes and a SIGWINCH was received.
             # We get 'Error: [Errno 4] Interrupted system call'
             # Just ignore.
             pass
 
-        # Wait for the real selector to be done.
-        th.join()
         assert result is not None
         return result
 
     def close(self) -> None:
         """
         Clean up resources.
         """
-        if self._r:
+        if self._r >= 0:
             os.close(self._r)
             os.close(self._w)
 

tests/test_inputhook.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+import time
+from types import SimpleNamespace
+
+import pytest
+
+import prompt_toolkit.application.application as application_module
+import prompt_toolkit.eventloop.inputhook as inputhook_module
+from prompt_toolkit.application.application import Application
+from prompt_toolkit.eventloop.inputhook import InputHookContext, InputHookSelector
+from prompt_toolkit.output import DummyOutput
+
+
+class StubSelector:
+    def __init__(self) -> None:
+        self.closed = False
+
+    def select(self, timeout: float | None = None) -> list[tuple[object, int]]:
+        return []
+
+    def close(self) -> None:
+        self.closed = True
+
+    def get_map(self) -> dict[object, object]:
+        return {}
+
+
+def test_inputhook_selector_handles_large_pipe_descriptors(monkeypatch):
+    writes = []
+    reads = []
+
+    def read(fd, size):
+        reads.append((fd, size))
+        return b"x"
+
+    fake_os = SimpleNamespace(
+        pipe=lambda: (1024, 1025),
+        write=lambda fd, data: writes.append((fd, data)) or len(data),
+        read=read,
+        close=lambda fd: None,
+    )
+    monkeypatch.setattr(inputhook_module, "os", fake_os)
+    monkeypatch.setattr(
+        inputhook_module,
+        "select",
+        SimpleNamespace(select=lambda *args: pytest.fail("select.select was called")),
+        raising=False,
+    )
+    monkeypatch.setattr(
+        inputhook_module, "get_running_loop", lambda: SimpleNamespace(_ready=[])
+    )
+
+    def inputhook(context: InputHookContext) -> None:
+        deadline = time.monotonic() + 1
+        while not context.input_is_ready():
+            if time.monotonic() >= deadline:
+                raise AssertionError("selector did not become ready")
+            time.sleep(0.001)
+
+    selector = InputHookSelector(StubSelector(), inputhook)
+
+    try:
+        assert selector.select(timeout=0) == []
+        assert writes == [(1025, b"x")]
+        assert reads == [(1024, 1024)]
+    finally:
+        selector.close()
+
+
+@pytest.mark.parametrize("pipe_fds", [(0, 1), (1024, 1025)])
+def test_inputhook_selector_close_handles_pipe_descriptors(monkeypatch, pipe_fds):
+    closed = []
+    fake_os = SimpleNamespace(pipe=lambda: pipe_fds, close=closed.append)
+    monkeypatch.setattr(inputhook_module, "os", fake_os)
+    selector = InputHookSelector(StubSelector(), lambda context: None)
+
+    selector.close()
+    selector.close()
+
+    assert closed == list(pipe_fds)
+
+
+@pytest.mark.parametrize("error", [KeyboardInterrupt, EOFError])
+def test_application_run_closes_inputhook_loop_after_interrupt(monkeypatch, error):
+    run_coro = object()
+    shutdown_coro = object()
+    calls = []
+
+    class StubLoop:
+        def run_until_complete(self, coro):
+            if coro is run_coro:
+                raise error
+            assert coro is shutdown_coro
+            calls.append("shutdown")
+
+        def shutdown_asyncgens(self):
+            return shutdown_coro
+
+        def close(self):
+            calls.append("close")
+
+    loop = StubLoop()
+    app = Application(output=DummyOutput())
+    app.run_async = lambda **kwargs: run_coro
+    monkeypatch.setattr(
+        application_module, "new_eventloop_with_inputhook", lambda inputhook: loop
+    )
+
+    with pytest.raises(error):
+        app.run(inputhook=lambda context: None)
+
+    assert calls == ["shutdown", "close"]