Feature: Use document extractors for grep and centralize file text extraction

open_terminal/main.py

@@ -503,22 +503,16 @@ async def read_file(
         mime, _ = mimetypes.guess_type(target)
         mime = mime or "application/octet-stream"
 
-        # Try document text extraction (PDF, Office, OpenDocument, etc.)
-        from open_terminal.utils.documents import EXTRACTORS
-
-        for ext_mime, ext_suffix, extractor in EXTRACTORS:
-            if (ext_mime and mime == ext_mime) or (
-                ext_suffix and target.lower().endswith(ext_suffix)
-            ):
-                text = await asyncio.to_thread(extractor, target)
-                lines = text.splitlines(keepends=True)
-                start = (start_line or 1) - 1
-                end = end_line or len(lines)
-                return {
-                    "path": target,
-                    "total_lines": len(lines),
-                    "content": "".join(lines[start:end]),
-                }
+        text = await asyncio.to_thread(_extract_text_with_supported_document_extractors, target, mime)
+        if text is not None:
+            lines = text.splitlines(keepends=True)
+            start = (start_line or 1) - 1
+            end = end_line or len(lines)
+            return {
+                "path": target,
+                "total_lines": len(lines),
+                "content": "".join(lines[start:end]),
+            }
 
         # Return raw binary for allowed mime type prefixes (e.g. image/*)
         if any(mime.startswith(prefix) for prefix in BINARY_FILE_MIME_PREFIXES):
@@ -539,6 +533,36 @@ async def read_file(
     }
 
 
+def _extract_text_with_supported_document_extractors(file_path: str, mime: str) -> str | None:
+    """Extract text for supported document types; return None when unsupported."""
+    from open_terminal.utils.documents import EXTRACTORS
+
+    for ext_mime, ext_suffix, extractor in EXTRACTORS:
+        if (ext_mime and mime == ext_mime) or (ext_suffix and file_path.lower().endswith(ext_suffix)):
+            return extractor(file_path)
+    return None
+
+
+def _read_file_as_text_representation_for_grep(file_path: str) -> str:
+    """Return searchable text using the same extraction behavior as read_file."""
+    try:
+        with open(file_path, "r", encoding="utf-8", errors="strict") as f:
+            return f.read()
+    except (UnicodeDecodeError, ValueError):
+        pass
+
+    import mimetypes
+
+    mime, _ = mimetypes.guess_type(file_path)
+    mime = mime or "application/octet-stream"
+
+    text = _extract_text_with_supported_document_extractors(file_path, mime)
+    if text is not None:
+        return text
+
+    raise UnicodeDecodeError("utf-8", b"", 0, 1, "Unsupported binary file")
+
+
 @app.get(
     "/files/display",
     operation_id="display_file",
@@ -749,7 +773,7 @@ async def replace_file_content(http_request: Request, request: ReplaceRequest, f
     "/files/grep",
     operation_id="grep_search",
     summary="Search file contents",
-    description="Search for a text pattern across files in a directory. Returns structured matches with file paths, line numbers, and matching lines. Skips binary files.",
+    description="Search for a text pattern across files in a directory. Returns structured matches with file paths, line numbers, and matching lines. Searches plain-text files directly and supported document binaries (PDF, Office, OpenDocument, etc.) via text extraction, using the same text-representation behavior as read_file. Unsupported binary files are skipped.",
     dependencies=[Depends(verify_api_key)],
     responses={
         404: {"description": "Search path not found."},
@@ -807,25 +831,25 @@ def _search_file(file_path: str):
             if truncated:
                 return
             try:
-                with open(file_path, "r", encoding="utf-8", errors="strict") as f:
-                    for line_number, line in enumerate(f, 1):
-                        if pattern.search(line):
-                            if match_per_line:
-                                matches.append(
-                                    {
-                                        "file": file_path,
-                                        "line": line_number,
-                                        "content": line.rstrip("\n\r"),
-                                    }
-                                )
-                                if len(matches) >= max_results:
-                                    truncated = True
-                                    return
-                            else:
-                                matches.append({"file": file_path})
-                                if len(matches) >= max_results:
-                                    truncated = True
-                                return  # one match per file is enough
+                content = _read_file_as_text_representation_for_grep(file_path)
+                for line_number, line in enumerate(content.splitlines(), 1):
+                    if pattern.search(line):
+                        if match_per_line:
+                            matches.append(
+                                {
+                                    "file": file_path,
+                                    "line": line_number,
+                                    "content": line.rstrip("\n\r"),
+                                }
+                            )
+                            if len(matches) >= max_results:
+                                truncated = True
+                                return
+                        else:
+                            matches.append({"file": file_path})
+                            if len(matches) >= max_results:
+                                truncated = True
+                            return  # one match per file is enough
             except (UnicodeDecodeError, ValueError, OSError):
                 pass  # skip binary or unreadable files
 
@@ -1404,11 +1428,6 @@ async def port_proxy(port: int, path: str, request: Request):
     from datetime import datetime as _datetime
     from fastapi.responses import JSONResponse
 
-    try:
-        import select as _select
-    except ImportError:
-        _select = None  # Not available on all platforms in all contexts
-
     # Determine terminal backend: prefer Unix PTY, then pywinpty, else None
     if _PTY_AVAILABLE:
         _TERMINAL_BACKEND = "pty"
@@ -1678,13 +1697,35 @@ async def ws_terminal(ws: WebSocket, session_id: str):
             master_fd = session["master_fd"]
             process = session["process"]
 
-            def _blocking_read():
-                """Read from PTY using select() so we don't block forever."""
+            async def _read_data():
+                """Read from the non-blocking PTY fd without occupying a worker thread."""
+                loop = asyncio.get_running_loop()
+
                 while not stop_event.is_set():
                     try:
-                        rlist, _, _ = _select.select([master_fd], [], [], 0.1)
-                        if rlist:
-                            return os.read(master_fd, 4096)
+                        return os.read(master_fd, 4096)
+                    except BlockingIOError:
+                        if process.poll() is not None:
+                            return b""
+
+                        future = loop.create_future()
+
+                        def _mark_readable():
+                            if not future.done():
+                                future.set_result(None)
+
+                        try:
+                            loop.add_reader(master_fd, _mark_readable)
+                        except (AttributeError, NotImplementedError):
+                            await asyncio.sleep(0.05)
+                            continue
+
+                        try:
+                            await asyncio.wait_for(future, timeout=0.1)
+                        except asyncio.TimeoutError:
+                            pass
+                        finally:
+                            loop.remove_reader(master_fd)
                     except (OSError, ValueError):
                         return b""
                 return b""
@@ -1715,6 +1756,9 @@ def _blocking_read():
                 except Exception:
                     return b""
 
+            async def _read_data():
+                return await loop.run_in_executor(None, _blocking_read)
+
             def _check_alive():
                 return pty_proc.isalive()
 
@@ -1730,7 +1774,7 @@ async def _pty_reader():
             """Forward PTY output -> WebSocket."""
             try:
                 while not stop_event.is_set():
-                    data = await loop.run_in_executor(None, _blocking_read)
+                    data = await _read_data()
                     if not data:
                         if stop_event.is_set():
                             break
@@ -1783,4 +1827,3 @@ async def _pty_reader():
     from open_terminal.utils.notebooks import create_notebooks_router
 
     app.include_router(create_notebooks_router(verify_api_key))
-

open_terminal/utils/runner.py

@@ -1,4 +1,5 @@
 import asyncio
+import errno
 import json
 import os
 import shlex
@@ -82,17 +83,59 @@ def __init__(self, command: str, cwd: str | None, env: dict | None, run_as_user:
             os.close(master_fd)
             raise
         os.close(slave_fd)
+
+        # The master side of a PTY is otherwise a blocking file descriptor.
+        # Reading it through the event loop's default executor ties up one
+        # worker thread for the lifetime of every running command, which can
+        # starve unrelated async file and subprocess operations under high
+        # concurrency.  Keep the fd non-blocking and wait for readability from
+        # the event loop instead.
+        flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
+        fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
         self._master_fd = master_fd
 
+    async def _wait_until_readable(self) -> None:
+        """Suspend until the PTY master fd becomes readable without a thread."""
+        loop = asyncio.get_running_loop()
+        future = loop.create_future()
+
+        def _mark_readable() -> None:
+            if not future.done():
+                future.set_result(None)
+
+        try:
+            loop.add_reader(self._master_fd, _mark_readable)
+        except (AttributeError, NotImplementedError):
+            # Some event loops do not expose fd readiness APIs.  This is still
+            # non-blocking: polling with a short sleep avoids occupying the
+            # default executor indefinitely.
+            await asyncio.sleep(0.05)
+            return
+
+        try:
+            await asyncio.wait_for(future, timeout=0.1)
+        except asyncio.TimeoutError:
+            pass
+        finally:
+            loop.remove_reader(self._master_fd)
+
     async def read_output(self, log_file) -> None:
-        loop = asyncio.get_event_loop()
         while True:
             try:
-                data = await loop.run_in_executor(None, os.read, self._master_fd, 4096)
-                if not data:
+                data = os.read(self._master_fd, 4096)
+            except BlockingIOError:
+                if self._process.poll() is not None:
                     break
-            except OSError:
-                break  # EIO when child exits
+                await self._wait_until_readable()
+                continue
+            except OSError as exc:
+                if exc.errno in (errno.EIO, errno.EBADF):
+                    break  # EIO is reported by PTYs when the child exits.
+                raise
+
+            if not data:
+                break
+
             if log_file:
                 await log_file.write(
                     json.dumps(

tests/test_runner.py

@@ -0,0 +1,38 @@
+import asyncio
+import os
+
+import pytest
+
+from open_terminal.utils.runner import PtyRunner, _PTY_AVAILABLE
+
+
+class MemoryLog:
+    def __init__(self):
+        self.records = []
+
+    async def write(self, data: str) -> None:
+        self.records.append(data)
+
+
+def test_pty_runner_reads_without_default_executor(monkeypatch):
+    if not _PTY_AVAILABLE:
+        pytest.skip("Unix PTY support is not available on this platform")
+
+    log = MemoryLog()
+    runner = PtyRunner("printf quick-win", cwd=None, env=os.environ.copy())
+
+    async def exercise_runner() -> None:
+        loop = asyncio.get_running_loop()
+
+        def fail_run_in_executor(*args, **kwargs):
+            raise AssertionError("PTY output reader should not use the default executor")
+
+        monkeypatch.setattr(loop, "run_in_executor", fail_run_in_executor)
+        await runner.read_output(log)
+
+    try:
+        asyncio.run(exercise_runner())
+        assert any("quick-win" in record for record in log.records)
+    finally:
+        runner._process.wait(timeout=5)
+        runner.close()