feat: add preview-mode ingest backend with live index proof (OSS + SaaS) (#2051)

* feat: add preview-mode ingest backend with live index proof (OSS + SaaS)
Introduce preview ingest plumbing so the frontend can show ingest progress
and verify indexed chunks without persisting Docling layout JSON.
- Add preview=true on upload router and connector sync (enabled in OSS and
  SaaS via is_ingest_preview_enabled; disabled on_prem)
- Thread preview_mode through UploadTask, task service, and connector sync
- Keep completed files in /tasks/enhanced for preview tasks (multi-file carousel)
- Add GET /ingest/preview/{task_id}/index-proof to query OpenSearch live for
  chunk metadata (phase, text previews, embedding model/dimensions)
- Track document_id on FileTask for index-proof lookups (set in Langflow and
  connector processors before ingest)
- Wire IngestPreviewService as a stateless helper (no in-memory Docling cache)
Intentionally excludes the spike's /docling endpoint and preview-only Docling
options (embedded page images). Layout skeleton stays frontend-only; backend
serves task phase + index proof only.
Tests: API guards, index proof service, router preview threading, enhanced
task serialization for preview vs normal tasks.

* Address code-review findings on the preview ingest backend:

- Pass the validated upload_task into IngestPreviewService instead of
  re-fetching from TaskService (closes preview_mode TOCTOU gap)
- Add service-level preview_mode guard and map not_preview_task to 404
- Gate index-proof endpoint with require_permission(knowledge:upload)
- Return per-request JSONResponse instances (no shared response object)
- Sort chunks by page + numeric suffix instead of lexicographic _id
- Use OpenSearch hits.total for chunk_count; expose chunks_returned and
  chunks_truncated when the 200-chunk cap applies
- Return file_not_found (404) for unknown ?file= paths in multi-file tasks
- Update Langflow processor comment for document_id threading
- Expand unit tests: single-fetch regression, ordering, totals, file_not_found,
  document_id threading, and non-preview rejection

* fix lint ruff

* adding a feature of flag

* address minor issues raised by Mike P

* style: ruff autofix (auto)

* ci: fail fast on runner/image architecture mismatch in integration suite

* ci: free up runner disk space before building/loading images

The move to ubuntu-latest runners (#2081) dropped available disk from the
~40GB self-hosted runners to ~14GB, so `docker load` of all four OpenRAG
images ran out of space unpacking OpenSearch ("no space left on device").

Add a step to build-images, test-suite, and e2e-run that removes the large
preinstalled toolchains these jobs never use (Android SDK, .NET, GHC,
CodeQL), reclaiming ~25GB before any image build/load.

Co-authored-by: Cursor <cursoragent@cursor.com>

* ci: scope image cache keys by runner architecture

The image cache key used only runner.os (always "Linux") plus a source
hash, with no CPU arch. After the ubuntu-latest migration (#2081) the amd64
build-images job hit the cache from earlier arm64 self-hosted runs and
restored/re-tagged stale arm64 images, shipping arm64 images to the amd64
test jobs (caught by the arch guard as a host/image mismatch).

Add runner.arch to every image cache key so amd64 and arm64 caches never
collide, forcing a native rebuild per architecture.

Co-authored-by: Cursor <cursoragent@cursor.com>

* ci: relocate Docker data-root to /mnt to fix disk exhaustion

Trimming preinstalled toolchains alone did not free enough space on the
small ubuntu-latest root filesystem (~30GB), so `docker load` of all four
OpenRAG images still hit "no space left on device". Move Docker's data-root
onto the runner's large ephemeral /mnt disk (~70GB) before building/loading
images, and additionally clear swift/powershell caches and apt lists.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix: forward explicit preview flag from v1 ingest endpoint

The public /v1/documents ingest endpoint calls upload_ingest_router directly
(not through FastAPI form parsing), so Form-defaulted params must be passed
explicitly. The new preview param was omitted, leaking the Form("false")
sentinel into `preview.lower()` and raising AttributeError -> HTTP 500 on
every SDK ingest/delete/filter call (18 sdk-python + 6 sdk-typescript
failures).

Pass preview="false" (preview is frontend-only; v1 SDK does not expose it)
and add a regression test asserting the flag is forwarded as a string.

Co-authored-by: Cursor <cursoragent@cursor.com>

* revert: drop CI runner/disk/cache workflow changes

Revert the arch-mismatch guard, disk cleanup, arch-scoped image cache keys,
and Docker data-root relocation added in this branch. The underlying CI
infrastructure issues were fixed separately in another PR, so restore
scripts/ci/run_integration_suite.sh and .github/workflows/test-ci.yml to
their pre-change state. Keeps the v1 preview ingest code fix.

Co-authored-by: Cursor <cursoragent@cursor.com>

* resolve comment 2a

---------

Co-authored-by: Olfa Maslah <olfamaslah@Olfas-MacBook-Pro.local>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 4 people

.env.example

@@ -411,6 +411,12 @@ SESSION_SECRET=
 # - "on_prem": Private Cloud
 OPENRAG_RUN_MODE=oss
 
+# Feature flag for the preview-mode ingest backend (default: false = disabled).
+# When true, the frontend can request preview ingests that show progress and a
+# live index proof without persisting Docling layout JSON. AND-ed with the run
+# mode: preview is only ever available in the "oss" and "saas" run modes above.
+# OPENRAG_INGEST_PREVIEW_ENABLED=false
+
 # Permission/identity cache backend. Only "memory" is currently wired.
 # When CACHE_BACKEND=memory the RBAC permission cache and OAuth-subject
 # cache are per-process, so OpenRAG must run with UVICORN_WORKERS=1 (and

src/api/connectors.py

@@ -28,6 +28,7 @@
     set_connector_access_bulk,
 )
 from session_manager import User
+from utils.ingest_preview_flag import is_ingest_preview_enabled
 from utils.logging_config import get_logger
 from utils.telemetry import Category, MessageId, TelemetryClient
 
@@ -592,6 +593,8 @@ class ConnectorSyncBody(BaseModel):
     # rather than failing. Set by the provider upload UI after the user confirms
     # overwrite in the duplicate dialog.
     replace_duplicates: bool = False
+    # When True (OSS/SaaS), run the ingest in preview mode (same as direct upload).
+    preview: bool = False
     # When True (COS only), index chunks without an owner field so OpenSearch DLS
     # makes them visible to all users in the instance. Temporary CIO mechanism;
     # not a full ACL feature. Defaults to False (private).
@@ -925,6 +928,10 @@ async def connector_sync(
             selected_files = [f.get("id") for f in selected_files_raw if f.get("id")]
             file_infos = selected_files_raw
 
+    # Preview mode is opt-in from the connector upload UI (OSS/SaaS). It applies
+    # to user-initiated ingests (explicit file selection), not automated re-sync.
+    preview_mode = body.preview and is_ingest_preview_enabled()
+
     try:
         await TelemetryClient.send_event(
             Category.CONNECTOR_OPERATIONS, MessageId.ORB_CONN_SYNC_START
@@ -1052,6 +1059,7 @@ async def connector_sync(
                 file_infos=file_infos,
                 ingest_settings=body.settings,
                 replace_duplicates=body.replace_duplicates,
+                preview_mode=preview_mode,
                 shared=body.shared,
             )
         elif body.sync_all or body.bucket_filter:
@@ -1156,6 +1164,7 @@ async def connector_sync(
                             new_ids,
                             jwt_token=jwt_token,
                             ingest_settings=body.settings,
+                            preview_mode=preview_mode,
                             shared=body.shared,
                         )
                     )
@@ -1168,6 +1177,7 @@ async def connector_sync(
                             jwt_token=jwt_token,
                             ingest_settings=body.settings,
                             replace_duplicates=True,
+                            preview_mode=preview_mode,
                             shared=body.shared,
                         )
                     )

src/api/ingest_preview.py

@@ -0,0 +1,67 @@
+"""Ephemeral ingest preview endpoints for preview-mode ingest."""
+
+from typing import Annotated, Any
+
+from fastapi import Depends
+from fastapi.responses import JSONResponse
+
+from dependencies import (
+    get_ingest_preview_service,
+    get_session_manager,
+    get_task_service,
+    require_permission,
+)
+from session_manager import User
+from utils.ingest_preview_flag import is_ingest_preview_enabled
+
+
+def _preview_unavailable_response() -> JSONResponse:
+    return JSONResponse(
+        {"error": "Ingest preview is not available in this run mode"},
+        status_code=404,
+    )
+
+
+def _require_preview_task(task_service: Any, user: User, task_id: str):
+    """Return (upload_task, error_response). Exactly one of the tuple values is None."""
+    if not is_ingest_preview_enabled():
+        return None, _preview_unavailable_response()
+
+    upload_task = task_service.get_upload_task(user.user_id, task_id)
+    if upload_task is None:
+        return None, JSONResponse({"error": "Task not found"}, status_code=404)
+    if not upload_task.preview_mode:
+        return None, JSONResponse({"error": "Task is not a preview ingest"}, status_code=404)
+    return upload_task, None
+
+
+async def get_index_proof(
+    task_id: str,
+    preview_service: Annotated[Any, Depends(get_ingest_preview_service)],
+    task_service: Annotated[Any, Depends(get_task_service)],
+    session_manager: Annotated[Any, Depends(get_session_manager)],
+    user: Annotated[User, Depends(require_permission("knowledge:upload"))],
+    file: str | None = None,
+):
+    """Return indexed chunk metadata proving embeddings landed in OpenSearch.
+
+    ``file`` selects a specific file within a multi-file preview task.
+    """
+    upload_task, error = _require_preview_task(task_service, user, task_id)
+    if error is not None:
+        return error
+
+    opensearch_client = session_manager.get_user_opensearch_client(user.user_id, user.jwt_token)
+    proof = await preview_service.get_index_proof(
+        upload_task=upload_task,
+        task_id=task_id,
+        opensearch_client=opensearch_client,
+        file_path=file,
+    )
+
+    if proof.get("error") == "not_preview_task":
+        return JSONResponse({"error": "Task is not a preview ingest"}, status_code=404)
+    if proof.get("error") == "file_not_found":
+        return JSONResponse({"error": "File not found in preview task"}, status_code=404)
+
+    return JSONResponse({"task_id": task_id, **proof})

src/api/router.py

@@ -17,6 +17,7 @@
     get_task_service,
 )
 from session_manager import User
+from utils.ingest_preview_flag import is_ingest_preview_enabled
 from utils.logging_config import get_logger
 
 logger = get_logger(__name__)
@@ -29,6 +30,7 @@ async def upload_ingest_router(
     tweaks_json: str | None = Form(None, alias="tweaks"),
     replace_duplicates: str = Form("true"),
     create_filter: str = Form("false"),
+    preview: str = Form("false"),
     document_service=Depends(get_document_service),
     langflow_file_service=Depends(get_langflow_file_service),
     session_manager=Depends(get_session_manager),
@@ -42,9 +44,14 @@ async def upload_ingest_router(
     - If DISABLE_INGEST_WITH_LANGFLOW is False (default): uses Langflow upload-ingest via task service
     """
     disable_ingest_with_langflow = get_openrag_config().knowledge.disable_ingest_with_langflow
+    requested_preview = preview.lower() == "true"
+    preview_mode = requested_preview and is_ingest_preview_enabled()
+    if requested_preview and not preview_mode:
+        logger.debug("Ingest preview ignored — not available in this run mode")
     logger.debug(
         "Router upload_ingest endpoint called",
         disable_langflow_ingest=disable_ingest_with_langflow,
+        preview_mode=preview_mode,
     )
 
     if disable_ingest_with_langflow:
@@ -53,6 +60,7 @@ async def upload_ingest_router(
             upload_files=file,
             replace_duplicates=replace_duplicates.lower() == "true",
             create_filter=create_filter.lower() == "true",
+            preview_mode=preview_mode,
             session_manager=session_manager,
             task_service=task_service,
             user=user,
@@ -67,6 +75,7 @@ async def upload_ingest_router(
         tweaks_json=tweaks_json,
         replace_duplicates=replace_duplicates.lower() == "true",
         create_filter=create_filter.lower() == "true",
+        preview_mode=preview_mode,
         langflow_file_service=langflow_file_service,
         session_manager=session_manager,
         task_service=task_service,
@@ -78,6 +87,7 @@ async def _traditional_upload_ingest_task(
     upload_files: list[UploadFile],
     replace_duplicates: bool,
     create_filter: bool,
+    preview_mode: bool,
     session_manager,
     task_service,
     user: User,
@@ -142,6 +152,7 @@ async def _traditional_upload_ingest_task(
                 original_filenames=file_path_to_original_filename,
                 replace_duplicates=replace_duplicates,
                 settings=settings,
+                preview_mode=preview_mode,
             )
 
             return JSONResponse(
@@ -151,6 +162,7 @@ async def _traditional_upload_ingest_task(
                     "file_count": len(upload_files),
                     "create_filter": create_filter,
                     "filename": original_filenames[0] if len(original_filenames) == 1 else None,
+                    "preview_mode": preview_mode,
                 },
                 status_code=202,
             )
@@ -180,6 +192,7 @@ async def _langflow_upload_ingest_task(
     tweaks_json,
     replace_duplicates: bool,
     create_filter: bool,
+    preview_mode: bool,
     langflow_file_service,
     session_manager,
     task_service,
@@ -251,6 +264,7 @@ async def _langflow_upload_ingest_task(
                 tweaks=tweaks,
                 settings=settings,
                 replace_duplicates=replace_duplicates,
+                preview_mode=preview_mode,
             )
 
             return JSONResponse(
@@ -260,6 +274,7 @@ async def _langflow_upload_ingest_task(
                     "file_count": len(upload_files),
                     "create_filter": create_filter,
                     "filename": original_filenames[0] if len(original_filenames) == 1 else None,
+                    "preview_mode": preview_mode,
                 },
                 status_code=202,
             )

src/api/v1/documents.py

@@ -69,6 +69,10 @@ async def ingest_endpoint(
         tweaks_json=tweaks,
         replace_duplicates=replace_duplicates,
         create_filter=create_filter,
+        # Preview ingest is a frontend-only feature; the v1 SDK does not expose
+        # it. Pass an explicit value so the Form("false") default sentinel is not
+        # forwarded when this function is called directly (not via form parsing).
+        preview="false",
         document_service=document_service,
         langflow_file_service=langflow_file_service,
         session_manager=session_manager,

src/app/container.py

@@ -25,6 +25,7 @@
 from services.document_service import DocumentService
 from services.flows_service import FlowsService
 from services.group_acl_service import GroupACLService
+from services.ingest_preview_service import IngestPreviewService
 from services.knowledge_filter_service import KnowledgeFilterService
 from services.langflow_file_service import LangflowFileService
 from services.langflow_ingest_token_service import LangflowIngestTokenService
@@ -82,6 +83,7 @@ async def initialize_services():
     models_service = ModelsService()
     document_index_writer = DocumentIndexWriter()
     langflow_ingest_token_service = LangflowIngestTokenService()
+    ingest_preview_service = IngestPreviewService()
     document_service = DocumentService(
         session_manager=session_manager,
         models_service=models_service,
@@ -232,6 +234,7 @@ def _lazy_session_factory():
         "langflow_mcp_service": langflow_mcp_service,
         "docling_service": clients.docling_service,
         "docling_polling_service": docling_polling_service,
+        "ingest_preview_service": ingest_preview_service,
         "rbac_service": rbac_service,
         "workspace_config_service": workspace_config_service,
     }

src/app/routes/internal.py

@@ -16,6 +16,7 @@
     documents,
     files,
     flows,
+    ingest_preview,
     knowledge_filter,
     langflow_files,
     langflow_ingest,
@@ -74,6 +75,14 @@ def register_internal_routes(app: FastAPI):
     app.add_api_route("/upload_options", upload.upload_options, methods=["GET"], tags=["internal"])
     app.add_api_route("/upload_bucket", upload.upload_bucket, methods=["POST"], tags=["internal"])
 
+    # Ingest preview endpoint (index proof for preview-mode ingests)
+    app.add_api_route(
+        "/ingest/preview/{task_id}/index-proof",
+        ingest_preview.get_index_proof,
+        methods=["GET"],
+        tags=["internal"],
+    )
+
     # Task endpoints
     # Literal sub-paths must be registered before the parameterised /{task_id}
     # so Starlette does not absorb "enhanced" as a task_id value.

src/config/settings.py

@@ -290,6 +290,18 @@ def is_azure_blob_enabled() -> bool:
     return raw in ("true", "1", "yes", "on")
 
 
+def is_ingest_preview_flag_enabled() -> bool:
+    """Raw opt-in flag for the preview-mode ingest backend.
+
+    Read per-call (like the other feature-flag accessors in this module) so
+    runtime/test overrides of ``OPENRAG_INGEST_PREVIEW_ENABLED`` take effect
+    without a restart. This is only the flag itself; run-mode gating is applied
+    by ``utils.ingest_preview_flag.is_ingest_preview_enabled()``.
+    """
+    raw = os.getenv("OPENRAG_INGEST_PREVIEW_ENABLED", "false").strip().lower()
+    return raw in ("true", "1", "yes", "on")
+
+
 def is_cloud_context() -> bool:
     """True when connector policy and SaaS settings guards should apply."""
     from utils.run_mode_utils import is_run_mode_saas

src/connectors/service.py

@@ -471,7 +471,10 @@ async def sync_connector_files(
 
         # Create custom task using TaskService
         task_id = await self.task_service.create_custom_task(
-            user_id, file_ids, processor, original_filenames=original_filenames
+            user_id,
+            file_ids,
+            processor,
+            original_filenames=original_filenames,
         )
 
         return task_id
@@ -485,6 +488,7 @@ async def sync_specific_files(
         file_infos: list[dict[str, Any]] = None,
         ingest_settings: dict[str, Any] | None = None,
         replace_duplicates: bool = False,
+        preview_mode: bool = False,
         shared: bool = False,
     ) -> str:
         """
@@ -662,7 +666,11 @@ async def sync_specific_files(
             }
 
         task_id = await self.task_service.create_custom_task(
-            user_id, expanded_file_ids, processor, original_filenames=original_filenames
+            user_id,
+            expanded_file_ids,
+            processor,
+            original_filenames=original_filenames,
+            preview_mode=preview_mode,
         )
 
         return task_id

src/dependencies.py

@@ -156,6 +156,10 @@ def get_langflow_file_service(services: dict = Depends(get_services)):
     return services["langflow_file_service"]
 
 
+def get_ingest_preview_service(services: dict = Depends(get_services)):
+    return services["ingest_preview_service"]
+
+
 def get_document_index_writer(services: dict = Depends(get_services)):
     return services["document_index_writer"]