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

Pass in-memory DocumentStream objects to both async Docling service operations so the original filename and complete file content are uploaded.

Fixes #6393

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

Author: Peter Gustafsson

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"