From c6f860b4ad7931478eacaac4d9ebff23d5893504 Mon Sep 17 00:00:00 2001 From: pfurovYnP <165936357+pfurovYnP@users.noreply.github.com> Date: Tue, 21 Apr 2026 11:21:54 +0300 Subject: [PATCH 1/3] Refactor shared document extraction logic for read and grep --- open_terminal/main.py | 95 +++++++++++++++++++++++++++---------------- 1 file changed, 59 insertions(+), 36 deletions(-) diff --git a/open_terminal/main.py b/open_terminal/main.py index c035741..1b59c7f 100644 --- a/open_terminal/main.py +++ b/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", @@ -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 @@ -1783,4 +1807,3 @@ async def _pty_reader(): from open_terminal.utils.notebooks import create_notebooks_router app.include_router(create_notebooks_router(verify_api_key)) - From c1b3ffa5903a6acf9b7ee43bd5da50e1dadb0a95 Mon Sep 17 00:00:00 2001 From: pfurovYnP <165936357+pfurovYnP@users.noreply.github.com> Date: Tue, 21 Apr 2026 11:43:30 +0300 Subject: [PATCH 2/3] update doc for grep --- open_terminal/main.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/open_terminal/main.py b/open_terminal/main.py index 1b59c7f..7825d11 100644 --- a/open_terminal/main.py +++ b/open_terminal/main.py @@ -773,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."}, From b0812cff34c5feb07c5726d53087dbfac5e65a74 Mon Sep 17 00:00:00 2001 From: pfurovYnP <165936357+pfurovYnP@users.noreply.github.com> Date: Wed, 3 Jun 2026 10:55:13 +0300 Subject: [PATCH 3/3] Avoid executor starvation for PTY reads --- open_terminal/main.py | 42 +++++++++++++++++++-------- open_terminal/utils/runner.py | 53 +++++++++++++++++++++++++++++++---- tests/test_runner.py | 38 +++++++++++++++++++++++++ 3 files changed, 117 insertions(+), 16 deletions(-) create mode 100644 tests/test_runner.py diff --git a/open_terminal/main.py b/open_terminal/main.py index 7825d11..116ed58 100644 --- a/open_terminal/main.py +++ b/open_terminal/main.py @@ -1428,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" @@ -1702,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"" @@ -1739,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() @@ -1754,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 diff --git a/open_terminal/utils/runner.py b/open_terminal/utils/runner.py index 2e63868..d01b5ab 100644 --- a/open_terminal/utils/runner.py +++ b/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( diff --git a/tests/test_runner.py b/tests/test_runner.py new file mode 100644 index 0000000..42297b3 --- /dev/null +++ b/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()