fix(file_processor): offload sync parsing to thread pool (#6095)

## Summary

- Inline file processors (pypdf, markitdown, docling) run CPU-intensive
parsing synchronously in the async event loop, blocking all concurrent
requests
- Wrap parsing calls in `asyncio.to_thread()` to offload to the thread
pool, matching the pattern used throughout the codebase (faiss,
sqlite-vec, milvus, localfs, sentence-transformers, etc.)

## Reproduction

With a ~4.5MB text file, concurrent `/v1/models` requests spike to
**6500ms+** latency during file processing (vs normal ~5ms). After fix,
max probe latency stays under **175ms**.

| | Before | After |
|---|---|---|
| Max probe latency | 6572ms | 173ms |
| Avg probe latency | 602ms | 9ms |

## Test plan

- [x] Existing unit tests pass (`tests/unit/providers/file_processor/` -
35 tests)
- [x] Pre-commit checks pass
- [x] Manual verification: server stays responsive during file
processing

Fixes #6094

Signed-off-by: Derek Higgins <derekh@redhat.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Charlie Doern <cdoern@redhat.com>

Author: 3 people

src/ogx/providers/inline/file_processor/docling/docling.py

@@ -4,8 +4,10 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
+import asyncio
 import os
 import tempfile
+import threading
 import time
 import uuid
 from typing import Any
@@ -49,6 +51,7 @@ def __init__(self, config: DoclingFileProcessorConfig, files_api=None) -> None:
         self.converter = DocumentConverter(
             format_options={InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options)}
         )
+        self._converter_lock = threading.Lock()
 
     async def process_file(
         self,
@@ -80,13 +83,25 @@ async def process_file(
             )
             content = content_response.body
 
+        return await asyncio.to_thread(self._process_content, content, filename, file_id, chunking_strategy, start_time)
+
+    def _process_content(
+        self,
+        content: bytes,
+        filename: str,
+        file_id: str | None,
+        chunking_strategy: VectorStoreChunkingStrategy | None,
+        start_time: float,
+    ) -> ProcessFileResponse:
+        """Convert and chunk file content. Runs in a thread."""
         # Preserve original file extension so DocumentConverter can detect the format
         suffix = os.path.splitext(filename)[1] or ".bin"
         with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp:
             tmp.write(content)
             tmp.flush()
 
-            result = self.converter.convert(tmp.name)
+            with self._converter_lock:
+                result = self.converter.convert(tmp.name)
 
         doc = result.document
         page_count = doc.num_pages()

src/ogx/providers/inline/file_processor/markitdown/markitdown_processor.py

@@ -4,8 +4,10 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
+import asyncio
 import os
 import tempfile
+import threading
 import time
 import uuid
 from typing import Any
@@ -40,6 +42,7 @@ def __init__(self, config: MarkItDownFileProcessorConfig, files_api) -> None:
         self.config = config
         self.files_api = files_api
         self.converter = MarkItDown()
+        self._converter_lock = threading.Lock()
 
     async def process_file(
         self,
@@ -69,13 +72,25 @@ async def process_file(
             )
             content = content_response.body
 
+        return await asyncio.to_thread(self._process_content, content, filename, file_id, chunking_strategy, start_time)
+
+    def _process_content(
+        self,
+        content: bytes,
+        filename: str,
+        file_id: str | None,
+        chunking_strategy: VectorStoreChunkingStrategy | None,
+        start_time: float,
+    ) -> ProcessFileResponse:
+        """Convert and chunk file content. Runs in a thread."""
         suffix = os.path.splitext(filename)[1] or ".bin"
         with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp:
             tmp.write(content)
             tmp.flush()
 
             try:
-                result = self.converter.convert(tmp.name)
+                with self._converter_lock:
+                    result = self.converter.convert(tmp.name)
             except Exception as e:
                 raise HTTPException(
                     status_code=422,

src/ogx/providers/inline/file_processor/pypdf/pypdf.py

@@ -4,6 +4,7 @@
 # This source code is licensed under the terms described in the LICENSE file in
 # the root directory of this source tree.
 
+import asyncio
 import io
 import mimetypes
 import time
@@ -74,9 +75,11 @@ async def process_file(
         mime_category = mime_type.split("/")[0] if (mime_type and "/" in mime_type) else None
 
         if mime_type == "application/pdf":
-            return self._process_pdf(content, filename, file_id, chunking_strategy, start_time)
+            return await asyncio.to_thread(self._process_pdf, content, filename, file_id, chunking_strategy, start_time)
         elif mime_category == "text":
-            return self._process_text(content, filename, file_id, chunking_strategy, start_time)
+            return await asyncio.to_thread(
+                self._process_text, content, filename, file_id, chunking_strategy, start_time
+            )
         else:
             raise HTTPException(
                 status_code=422,