fix(file-processor): Preserve content in async Docling uploads (#6394)

# What does this PR do?

This fixes async uploads to Docling Serve.

OGX wrote the document to a temporary file and sent its path before the
data was flushed. Small documents could therefore be uploaded as empty
files. The temporary filename also hid the original filename and file
type.

This change sends the document as an in-memory DocumentStream instead.
It keeps the original filename and sends all file bytes. The fix is used
for both async conversion and async chunking.

Fixes #6393

## Test Plan

- Run: uv run pytest -q
tests/unit/providers/file_processor/test_docling_serve.py
- Result: 23 passed.
- Run pre-commit on the two changed files.
- Result: all checks passed, including Ruff and mypy.
- Run all file-processor unit tests.
- Result: 87 passed and 2 skipped. Four unrelated auto-processor tests
failed because the local environment detected their fake DOCX and XLSX
data as an unknown MIME type.

The full pre-commit run was also checked. All applicable checks passed
except two API coverage hooks because oasdiff is not installed locally.

Signed-off-by: Peter Gustafsson <peter.gustafsson6@gmail.com>
Co-authored-by: Peter Gustafsson <peter.gustafsson6@gmail.com>

Author: pgustafs

src/ogx/providers/remote/file_processor/docling_serve/docling_serve.py

@@ -5,16 +5,16 @@
 # the root directory of this source tree.
 
 import os
-import tempfile
 import time
 import uuid
-from pathlib import Path
+from io import BytesIO
 from typing import Any
 
 import httpx
 from docling.datamodel.base_models import OutputFormat
 from docling.datamodel.service.options import ConvertDocumentsOptions
 from docling.service_client import AsyncDoclingServiceClient, ChunkerKind
+from docling_core.types.io import DocumentStream
 from fastapi import UploadFile
 
 from ogx.log import get_logger
@@ -222,37 +222,33 @@ async def _convert_no_chunk_async(
         document_metadata: dict[str, Any],
     ) -> list[Chunk]:
         """Convert file using async endpoints with AsyncDoclingServiceClient."""
-        # AsyncDoclingServiceClient requires a file path via temp file
-        with tempfile.NamedTemporaryFile() as tmp:
-            tmp.write(content)
-            tmp_path = Path(tmp.name)
-
-            async with AsyncDoclingServiceClient(
-                url=self.config.base_url,
-                api_key=self.config.api_key.get_secret_value() if self.config.api_key else "",
-                job_timeout=300.0,
-            ) as client:
-                job = await client.submit(
-                    source=tmp_path,
-                    options=ConvertDocumentsOptions(to_formats=[OutputFormat.MARKDOWN]),
-                )
-                result = await job.result()
-
-            # Handle both local docling-serve (ConversionResult with .document)
-            # and IBM SaaS (PresignedUrlConvertResponse with .documents and presigned URLs)
-            md_content = ""
-            if hasattr(result, "documents"):
-                # IBM SaaS: PresignedUrlConvertResponse with presigned URLs
-                if result.documents and result.documents[0].artifacts:
-                    artifact = result.documents[0].artifacts[0]
-                    # Download markdown from presigned URL
-                    async with httpx.AsyncClient() as http_client:
-                        response = await http_client.get(str(artifact.uri))
-                        response.raise_for_status()
-                        md_content = response.text
-            elif hasattr(result, "document"):
-                # Local docling-serve: ConversionResult with direct document
-                md_content = result.document.export_to_markdown() if result.document else ""
+        source = DocumentStream(name=filename, stream=BytesIO(content))
+        async with AsyncDoclingServiceClient(
+            url=self.config.base_url,
+            api_key=self.config.api_key.get_secret_value() if self.config.api_key else "",
+            job_timeout=300.0,
+        ) as client:
+            job = await client.submit(
+                source=source,
+                options=ConvertDocumentsOptions(to_formats=[OutputFormat.MARKDOWN]),
+            )
+            result = await job.result()
+
+        # Handle both local docling-serve (ConversionResult with .document)
+        # and IBM SaaS (PresignedUrlConvertResponse with .documents and presigned URLs)
+        md_content = ""
+        if hasattr(result, "documents"):
+            # IBM SaaS: PresignedUrlConvertResponse with presigned URLs
+            if result.documents and result.documents[0].artifacts:
+                artifact = result.documents[0].artifacts[0]
+                # Download markdown from presigned URL
+                async with httpx.AsyncClient() as http_client:
+                    response = await http_client.get(str(artifact.uri))
+                    response.raise_for_status()
+                    md_content = response.text
+        elif hasattr(result, "document"):
+            # Local docling-serve: ConversionResult with direct document
+            md_content = result.document.export_to_markdown() if result.document else ""
 
         if not md_content or not md_content.strip():
             return []
@@ -374,39 +370,35 @@ async def _convert_and_chunk_async(
         document_metadata: dict[str, Any],
     ) -> list[Chunk]:
         """Convert and chunk file using async endpoints with AsyncDoclingServiceClient."""
-        # AsyncDoclingServiceClient requires a file path via temp file
-        with tempfile.NamedTemporaryFile() as tmp:
-            tmp.write(content)
-            tmp_path = Path(tmp.name)
-
-            async with AsyncDoclingServiceClient(
-                url=self.config.base_url,
-                api_key=self.config.api_key.get_secret_value() if self.config.api_key else "",
-                job_timeout=300.0,
-            ) as client:
-                try:
-                    job = await client.submit_chunk(
-                        source=tmp_path,
-                        chunker=ChunkerKind.HYBRID,
-                        options=ConvertDocumentsOptions(),
-                    )
-                    response = await job.result()
-                except httpx.HTTPStatusError as e:
-                    # Chunking endpoint not supported (e.g., IBM Docling SaaS)
-                    if e.response.status_code in (404, 405):
-                        raise InvalidParameterError(
-                            param_name="chunking_strategy",
-                            value=chunking_strategy.model_dump() if chunking_strategy else None,
-                            constraint=(
-                                "Chunking is not supported by this Docling instance. "
-                                "This is a known limitation of IBM Docling SaaS. "
-                                "Either remove 'chunking_strategy' from your request, "
-                                "or configure OGX to use local docling-serve for chunking support."
-                            ),
-                        ) from e
-                    raise
+        source = DocumentStream(name=filename, stream=BytesIO(content))
+        async with AsyncDoclingServiceClient(
+            url=self.config.base_url,
+            api_key=self.config.api_key.get_secret_value() if self.config.api_key else "",
+            job_timeout=300.0,
+        ) as client:
+            try:
+                job = await client.submit_chunk(
+                    source=source,
+                    chunker=ChunkerKind.HYBRID,
+                    options=ConvertDocumentsOptions(),
+                )
+                response = await job.result()
+            except httpx.HTTPStatusError as e:
+                # Chunking endpoint not supported (e.g., IBM Docling SaaS)
+                if e.response.status_code in (404, 405):
+                    raise InvalidParameterError(
+                        param_name="chunking_strategy",
+                        value=chunking_strategy.model_dump() if chunking_strategy else None,
+                        constraint=(
+                            "Chunking is not supported by this Docling instance. "
+                            "This is a known limitation of IBM Docling SaaS. "
+                            "Either remove 'chunking_strategy' from your request, "
+                            "or configure OGX to use local docling-serve for chunking support."
+                        ),
+                    ) from e
+                raise
 
-            raw_chunks = response.chunks if response.chunks else []
+        raw_chunks = response.chunks if response.chunks else []
 
         if not raw_chunks:
             return []

tests/unit/providers/file_processor/test_docling_serve.py

@@ -11,6 +11,7 @@
 
 import httpx
 import pytest
+from docling_core.types.io import DocumentStream
 from fastapi import UploadFile
 from pydantic import SecretStr
 
@@ -448,6 +449,10 @@ async def test_ibm_saas_allows_conversion_without_chunking(
                 assert result.chunks is not None
                 assert len(result.chunks) > 0
                 assert result.metadata["conversion_method"] == "async"
+                source = mock_instance.submit.await_args.kwargs["source"]
+                assert isinstance(source, DocumentStream)
+                assert source.name == "test.pdf"
+                assert source.stream.getvalue() == b"%PDF-fake-content"
 
     async def test_local_docker_allows_chunking(self, upload_file: UploadFile):
         """Local docling-serve should allow chunking (successful response)."""
@@ -481,6 +486,11 @@ async def test_local_docker_allows_chunking(self, upload_file: UploadFile):
             # Should succeed without raising InvalidParameterError
             result = await processor.process_file(request, file=upload_file)
 
+            submit_kwargs = mock_instance.submit_chunk.await_args.kwargs
+            source = submit_kwargs["source"]
+            assert isinstance(source, DocumentStream)
+            assert source.name == "test.pdf"
+            assert source.stream.getvalue() == b"%PDF-fake-content"
         assert result.chunks is not None
         assert len(result.chunks) > 0
         assert result.metadata["conversion_method"] == "async"