feat: Implement two-phase ingestion flow with Docling polling

src/app/container.py

@@ -5,6 +5,7 @@
 
 from api.connector_router import ConnectorRouter
 from config.settings import (
+    ENABLE_BACKEND_DOCLING_POLLING,
     INGESTION_TIMEOUT,
     JWT_SIGNING_KEY,
     SESSION_SECRET,
@@ -19,6 +20,7 @@
 from services.api_key_service import APIKeyService
 from services.auth_service import AuthService
 from services.chat_service import ChatService
+from services.docling_polling_service import DoclingPollingService
 from services.document_service import DocumentService
 from services.flows_service import FlowsService
 from services.knowledge_filter_service import KnowledgeFilterService
@@ -69,11 +71,24 @@ async def initialize_services():
     )
     search_service = SearchService(session_manager, models_service)
     register_search_service(search_service)
+
+    # Backend-side Docling polling coordinator. Constructed once as a
+    # singleton (it is stateless) and gated by ENABLE_BACKEND_DOCLING_POLLING
+    # so operators can roll back to the legacy single-call ingestion path
+    # without code changes. When disabled, downstream callers receive None
+    # and fall through to the legacy flow.
+    docling_polling_service = (
+        DoclingPollingService(clients.docling_service)
+        if ENABLE_BACKEND_DOCLING_POLLING and clients.docling_service is not None
+        else None
+    )
+
     task_service = TaskService(
         document_service,
         models_service,
         ingestion_timeout=INGESTION_TIMEOUT,
         docling_service=clients.docling_service,
+        docling_polling_service=docling_polling_service,
     )
     flows_service = FlowsService()
     chat_service = ChatService(flows_service=flows_service)
@@ -197,6 +212,7 @@ def _lazy_session_factory():
         "api_key_service": api_key_service,
         "langflow_mcp_service": langflow_mcp_service,
         "docling_service": clients.docling_service,
+        "docling_polling_service": docling_polling_service,
         "rbac_service": rbac_service,
         "workspace_config_service": workspace_config_service,
     }

src/config/settings.py

@@ -161,6 +161,22 @@
 # Default: 3600 seconds (60 minutes)
 INGESTION_TIMEOUT = get_env_int("INGESTION_TIMEOUT", 3600)
 
+# Two-phase ingestion: backend-side Docling polling configuration.
+# Controls how the OpenRAG backend waits for Docling Serve to finish converting
+# a document before invoking the Langflow ingestion flow. Decoupling this poll
+# from Langflow keeps Langflow execution slots free during long Docling jobs.
+# When ENABLE_BACKEND_DOCLING_POLLING is false, the backend submits to Docling
+# and immediately invokes Langflow with the task_id; Langflow's DoclingRemote
+# component then polls Docling itself (legacy single-call behavior).
+ENABLE_BACKEND_DOCLING_POLLING = os.getenv(
+    "ENABLE_BACKEND_DOCLING_POLLING", "true"
+).lower() in ("true", "1", "yes")
+DOCLING_POLL_INTERVAL_SECONDS = get_env_float("DOCLING_POLL_INTERVAL_SECONDS", 3.0)
+DOCLING_POLL_MAX_SECONDS = get_env_int("DOCLING_POLL_MAX_SECONDS", 1800)
+DOCLING_POLL_MAX_INTERVAL_SECONDS = get_env_float("DOCLING_POLL_MAX_INTERVAL_SECONDS", 30.0)
+DOCLING_POLL_BACKOFF_FACTOR = get_env_float("DOCLING_POLL_BACKOFF_FACTOR", 1.5)
+DOCLING_POLL_TRANSIENT_RETRIES = get_env_int("DOCLING_POLL_TRANSIENT_RETRIES", 5)
+
 def is_no_auth_mode():
     """Check if we're running in no-auth mode (OAuth credentials missing)"""
     if IBM_AUTH_ENABLED:

src/dependencies.py

@@ -17,7 +17,7 @@
 
 import asyncio
 import dataclasses
 from typing import AsyncIterator, Optional, Sequence
 
 from cachetools import TTLCache
 from fastapi import Depends, HTTPException, Request
@@ -76,39 +76,39 @@
     return request.app.state.services
 
 
 def get_session_manager(services: dict = Depends(get_services)):
     return services["session_manager"]
 
 
 def get_auth_service(services: dict = Depends(get_services)):
     return services["auth_service"]
 
 
 def get_chat_service(services: dict = Depends(get_services)):
     return services["chat_service"]
 
 
 def get_search_service(services: dict = Depends(get_services)):
     return services["search_service"]
 
 
 def get_document_service(services: dict = Depends(get_services)):
     return services["document_service"]
 
 
 def get_task_service(services: dict = Depends(get_services)):
     return services["task_service"]
 
 
 def get_knowledge_filter_service(services: dict = Depends(get_services)):
     return services["knowledge_filter_service"]
 
 
 def get_monitor_service(services: dict = Depends(get_services)):
     return services["monitor_service"]
 
 
 def get_connector_service(services: dict = Depends(get_services)):
     return services["connector_service"]
 
 
@@ -130,6 +130,10 @@
 
 def get_docling_service(services: dict = Depends(get_services)):
     return services["docling_service"]
+
+
+def get_docling_polling_service(services: dict = Depends(get_services)):
+    return services["docling_polling_service"]
 def get_rbac_service(services: dict = Depends(get_services)):
     return services["rbac_service"]
 

src/models/processors.py

@@ -806,7 +806,7 @@ async def process_item(
 
 
 class LangflowFileProcessor(TaskProcessor):
-    """Processor for Langflow file uploads with upload and ingest"""
+    """Processor for Langflow file uploads with two-phase Docling + Langflow ingestion."""
 
     def __init__(
         self,
@@ -821,6 +821,7 @@ def __init__(
         settings: dict = None,
         replace_duplicates: bool = False,
         connector_type: str = "local",
+        docling_polling_service=None,
     ):
         super().__init__()
         self.langflow_file_service = langflow_file_service
@@ -834,6 +835,11 @@ def __init__(
         self.settings = settings
         self.replace_duplicates = replace_duplicates
         self.connector_type = connector_type
+        # Backend-side Docling polling coordinator. Injected by TaskService
+        # from the container; gating by ENABLE_BACKEND_DOCLING_POLLING happens
+        # at construction time in app.container. When None, the legacy
+        # single-call ingestion path is used.
+        self.docling_polling_service = docling_polling_service
 
     async def process_item(
         self, upload_task: UploadTask, item: str, file_task: FileTask
@@ -908,7 +914,10 @@ async def process_item(
             # Prepare metadata tweaks similar to API endpoint
             final_tweaks = self.tweaks.copy() if self.tweaks else {}
 
-            # Process file using langflow service
+            # Process file using langflow service. Passing the polling
+            # service triggers the two-phase model: backend polls Docling,
+            # then invokes Langflow only after SUCCESS. file_task is passed
+            # so phase / docling_status are tracked on the task record.
             result = await self.langflow_file_service.upload_and_ingest_file(
                 file_tuple=file_tuple,
                 session_id=self.session_id,
@@ -919,6 +928,8 @@ async def process_item(
                 owner_name=self.owner_name,
                 owner_email=self.owner_email,
                 connector_type=self.connector_type,
+                docling_polling_service=self.docling_polling_service,
+                file_task=file_task,
             )
 
             # Update task with success
@@ -930,7 +941,7 @@ async def process_item(
         except Exception as e:
             # Update task with failure
             file_task.status = TaskStatus.FAILED
-            file_task.error_message = str(e)
+            file_task.error = str(e)
             file_task.updated_at = time.time()
             upload_task.failed_files += 1
             raise

src/models/tasks.py

@@ -12,6 +12,28 @@ class TaskStatus(Enum):
     FAILED = "failed"
 
 
+class DoclingPhaseStatus(Enum):
+    """Tracks the state of the Docling conversion sub-phase for a single file."""
+    PENDING = "pending"
+    PROCESSING = "processing"
+    SUCCESS = "success"
+    FAILED = "failed"
+    EXPIRED = "expired"
+
+
+class IngestionPhase(Enum):
+    """Tracks which phase of the two-phase ingestion pipeline a file is in.
+
+    DOCLING: file has been submitted to Docling Serve and the backend is
+        polling for conversion completion. Langflow has not been called.
+    LANGFLOW: Docling conversion succeeded; Langflow ingestion flow is running.
+    COMPLETE: Langflow ingestion finished and the document is indexed.
+    """
+    DOCLING = "docling"
+    LANGFLOW = "langflow"
+    COMPLETE = "complete"
+
+
 @dataclass
 class FileTask:
     file_path: str
@@ -22,6 +44,11 @@ class FileTask:
     created_at: float = field(default_factory=time.time)
     updated_at: float = field(default_factory=time.time)
     filename: Optional[str] = None  # Original filename for display
+    # Two-phase ingestion fields. Only meaningful for processors that submit
+    # files to Docling Serve and then trigger Langflow (i.e. LangflowFileProcessor).
+    docling_task_id: Optional[str] = None
+    docling_status: DoclingPhaseStatus = DoclingPhaseStatus.PENDING
+    phase: IngestionPhase = IngestionPhase.DOCLING
 
     @property
     def duration_seconds(self) -> float:

src/services/docling_polling_service.py

@@ -0,0 +1,152 @@
+"""Backend-side Docling polling coordinator.
+
+Owns the wait loop that previously lived inside the Langflow ingestion flow's
+DoclingRemote component. Keeping the loop here means a Langflow execution slot
+is held only for the brief chunk/embed/index phase, not for the entire Docling
+conversion (which can be many minutes for large or OCR-heavy documents).
+"""
+
+import asyncio
+import time
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional
+
+from services.docling_service import (
+    DoclingService,
+    DoclingStatusSnapshot,
+    DoclingTaskState,
+)
+from utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class PollOutcome(str, Enum):
+    SUCCESS = "success"
+    FAILED = "failed"
+    EXPIRED = "expired"
+    TIMEOUT = "timeout"
+
+
+@dataclass
+class DoclingPollResult:
+    outcome: PollOutcome
+    detail: Optional[str] = None
+    last_snapshot: Optional[DoclingStatusSnapshot] = None
+    elapsed_seconds: float = 0.0
+
+
+class DoclingPollingService:
+    """Polls a Docling task to terminal state without involving Langflow."""
+
+    def __init__(self, docling_service: DoclingService):
+        self.docling_service = docling_service
+
+    async def poll_until_ready(
+        self,
+        task_id: str,
+        poll_interval: float,
+        max_seconds: float,
+        max_interval: float = 30.0,
+        backoff_factor: float = 1.5,
+        transient_retry_budget: int = 5,
+    ) -> DoclingPollResult:
+        """Loop on Docling status until terminal or until max_seconds elapses.
+
+        Transient errors (network, 5xx, NOT_FOUND seen briefly before the task
+        is registered server-side) are absorbed up to ``transient_retry_budget``
+        before being surfaced as failures. The interval grows by
+        ``backoff_factor`` after each non-success poll, capped at
+        ``max_interval``, so we don't hammer Docling for slow conversions.
+        """
+        if poll_interval <= 0:
+            raise ValueError("poll_interval must be > 0")
+        if max_seconds <= 0:
+            raise ValueError("max_seconds must be > 0")
+
+        start = time.monotonic()
+        deadline = start + max_seconds
+        interval = poll_interval
+        consecutive_not_found = 0
+        last_snapshot: Optional[DoclingStatusSnapshot] = None
+
+        logger.debug(
+            "Starting Docling polling",
+            task_id=task_id)
+
+        while True:
+            logger.debug(
+            "Docling polling",
+            task_id=task_id)
+            snapshot = await self.docling_service.check_task_status(task_id)
+            last_snapshot = snapshot
+            logger.debug(
+                "Snapshot received",
+                task_id=task_id,
+                snapshot=last_snapshot
+            )
+            elapsed = time.monotonic() - start
+
+            if snapshot.state == DoclingTaskState.SUCCESS:
+                logger.debug(
+                    "Docling task reached SUCCESS",
+                    task_id=task_id,
+                    elapsed_seconds=round(elapsed, 2),
+                )
+                return DoclingPollResult(
+                    outcome=PollOutcome.SUCCESS,
+                    last_snapshot=snapshot,
+                    elapsed_seconds=elapsed,
+                )
+
+            if snapshot.state == DoclingTaskState.FAILED:
+                logger.warning(
+                    "Docling task reported FAILED",
+                    task_id=task_id,
+                    detail=snapshot.detail,
+                    elapsed_seconds=round(elapsed, 2),
+                )
+                return DoclingPollResult(
+                    outcome=PollOutcome.FAILED,
+                    detail=snapshot.detail or "Docling reported failure",
+                    last_snapshot=snapshot,
+                    elapsed_seconds=elapsed,
+                )
+
+            if snapshot.state == DoclingTaskState.NOT_FOUND:
+                # NOT_FOUND immediately after submission can be a brief window
+                # before the task is visible. Tolerate up to the transient
+                # budget; beyond that, treat as expired/unknown.
+                consecutive_not_found += 1
+                if consecutive_not_found > transient_retry_budget:
+                    logger.warning(
+                        "Docling task NOT_FOUND past retry budget",
+                        task_id=task_id,
+                        elapsed_seconds=round(elapsed, 2),
+                    )
+                    return DoclingPollResult(
+                        outcome=PollOutcome.EXPIRED,
+                        detail="Docling task not found (expired or unknown task_id)",
+                        last_snapshot=snapshot,
+                        elapsed_seconds=elapsed,
+                    )
+            else:
+                consecutive_not_found = 0
+
+            # Compute remaining time before deadline; sleep at most that long.
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                logger.warning(
+                    "Docling polling exceeded max_seconds",
+                    task_id=task_id,
+                    max_seconds=max_seconds,
+                )
+                return DoclingPollResult(
+                    outcome=PollOutcome.TIMEOUT,
+                    detail=f"Docling polling timed out after {max_seconds}s",
+                    last_snapshot=snapshot,
+                    elapsed_seconds=time.monotonic() - start,
+                )
+            await asyncio.sleep(min(interval, remaining))
+            interval = min(interval * backoff_factor, max_interval)