feat: disable langflow ingestion via UI (#1623)

* Added disable langflow ingestion setting on backend

* Added option to disable langflow ingestion in frontend

* Fixed ingestion not going to tasks

* style: ruff format (auto)

* fix frontend lint errors

* fixed type errors

* Potential fix for pull request finding 'CodeQL / Information exposure through an exception'

Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>

* fixed mypy findings

* fixed lint

* fix: apply CodeRabbit auto-fixes

Fixed 4 file(s) based on 6 unresolved review comments.

Co-authored-by: CodeRabbit <noreply@coderabbit.ai>

* fix: apply CodeRabbit auto-fixes

Fixed 2 file(s) based on 1 unresolved review comment.

Co-authored-by: CodeRabbit <noreply@coderabbit.ai>

* style: ruff format (auto)

* fix mypi

* added acl support

* style: ruff format (auto)

* fix mypy

* fixed integration tests

* pass temp file paths as empty to not delete default files on error

* wait for opensearch

* fix session manager jwt

* fix lint

* fix spaces

* fix health check

* fix kwargs

* style: ruff format (auto)

* fixed nitpicks from coderabbit

* style: ruff format (auto)

* fix ruff

* fix result

* fix types

* Match connector_file_id in ACL/metadata updates (#1704)

Ensure ACL and metadata updates target both Langflow and non-Langflow chunks by matching document IDs against multiple fields. Introduce _build_id_query and id_fields parameters to should_update_acl, update_document_acl, and batch_update_acls so updates can match document_id and connector_file_id. Stop overwriting chunk owner during ACL syncs (only allowed_users/allowed_groups are updated) and set owner at ingest to the syncing user in TaskProcessor. Adjust tests to expect the combined bool query and make minor typing/import/formatting cleanups.

* mypy fix

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <62310815+github-advanced-security[bot]@users.noreply.github.com>
Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>
Co-authored-by: CodeRabbit <noreply@coderabbit.ai>
Co-authored-by: Edwin Jose <edwin.jose@datastax.com>
Co-authored-by: Edwin Jose <edwinjose900@gmail.com>

Author: 6 people

frontend/app/api/mutations/useUpdateSettingsMutation.ts

@@ -18,6 +18,7 @@ export interface UpdateSettingsRequest {
   table_structure?: boolean;
   ocr?: boolean;
   picture_descriptions?: boolean;
+  disable_ingest_with_langflow?: boolean;
   embedding_model?: string;
   embedding_provider?: string;
 

frontend/app/api/queries/useGetConversationsQuery.ts

@@ -48,7 +48,10 @@ export interface ConversationHistoryResponse {
 export const useGetConversationsQuery = (
   endpoint: EndpointType,
   refreshTrigger?: number,
-  options?: Omit<UseQueryOptions, "queryKey" | "queryFn">,
+  options?: Omit<
+    UseQueryOptions<ChatConversation[], Error, ChatConversation[]>,
+    "queryKey" | "queryFn"
+  >,
 ) => {
   const queryClient = useQueryClient();
   const { isOnboardingComplete } = useChat();

frontend/app/api/queries/useGetNudgesQuery.ts

@@ -25,7 +25,10 @@ export interface NudgeQueryParams {
 
 export const useGetNudgesQuery = (
   params: NudgeQueryParams | null = {},
-  options?: Omit<UseQueryOptions, "queryKey" | "queryFn">,
+  options?: Omit<
+    UseQueryOptions<Nudge[], Error, Nudge[]>,
+    "queryKey" | "queryFn"
+  >,
 ) => {
   const { chatId, filters, limit, scoreThreshold } = params ?? {};
   const queryClient = useQueryClient();

frontend/app/api/queries/useGetSettingsQuery.ts

@@ -18,6 +18,7 @@ export interface KnowledgeSettings {
   table_structure?: boolean;
   ocr?: boolean;
   picture_descriptions?: boolean;
+  disable_ingest_with_langflow?: boolean;
 }
 
 export interface ProviderSettings {

frontend/app/settings/_components/ingest-settings-section.tsx

@@ -46,6 +46,8 @@ export function IngestSettingsSection() {
   const [ocr, setOcr] = useState<boolean>(false);
   const [pictureDescriptions, setPictureDescriptions] =
     useState<boolean>(false);
+  const [disableIngestWithLangflow, setDisableIngestWithLangflow] =
+    useState<boolean>(false);
 
   const { data: settings = {} } = useGetSettingsQuery({
     enabled: isAuthenticated || isNoAuthMode,
@@ -146,13 +148,22 @@ export function IngestSettingsSection() {
       setPictureDescriptions(settings.knowledge.picture_descriptions);
   }, [settings.knowledge?.picture_descriptions]);
 
+  useEffect(() => {
+    if (settings.knowledge?.disable_ingest_with_langflow !== undefined)
+      setDisableIngestWithLangflow(
+        settings.knowledge.disable_ingest_with_langflow,
+      );
+  }, [settings.knowledge?.disable_ingest_with_langflow]);
+
   const k = settings.knowledge;
   const knowledgeIngestDirty =
     chunkSize !== (k?.chunk_size ?? chunkSize) ||
     chunkOverlap !== (k?.chunk_overlap ?? chunkOverlap) ||
     tableStructure !== (k?.table_structure ?? tableStructure) ||
     ocr !== (k?.ocr ?? ocr) ||
-    pictureDescriptions !== (k?.picture_descriptions ?? pictureDescriptions);
+    pictureDescriptions !== (k?.picture_descriptions ?? pictureDescriptions) ||
+    disableIngestWithLangflow !==
+      (k?.disable_ingest_with_langflow ?? disableIngestWithLangflow);
 
   const handleEmbeddingModelChange = (newModel: string, provider?: string) => {
     if (newModel && provider) {
@@ -195,6 +206,7 @@ export function IngestSettingsSection() {
         table_structure: tableStructure,
         ocr,
         picture_descriptions: pictureDescriptions,
+        disable_ingest_with_langflow: disableIngestWithLangflow,
       },
       { onSuccess: () => setChunkValidationError(null) },
     );
@@ -238,6 +250,7 @@ export function IngestSettingsSection() {
         setTableStructure(DEFAULT_KNOWLEDGE_SETTINGS.table_structure);
         setOcr(DEFAULT_KNOWLEDGE_SETTINGS.ocr);
         setPictureDescriptions(DEFAULT_KNOWLEDGE_SETTINGS.picture_descriptions);
+        setDisableIngestWithLangflow(false);
         setChunkValidationError(null);
         closeDialog();
       })
@@ -429,6 +442,25 @@ export function IngestSettingsSection() {
             </div>
           </div>
           <div>
+            <div className="flex items-center justify-between py-3 border-b border-border">
+              <div className="flex-1">
+                <Label
+                  htmlFor="disable-ingest-with-langflow"
+                  className="text-base font-medium cursor-pointer pb-3"
+                >
+                  Disable Langflow Ingestion
+                </Label>
+                <div className="text-sm text-muted-foreground">
+                  Bypass Langflow for document ingestion and use traditional
+                  processing.
+                </div>
+              </div>
+              <Switch
+                id="disable-ingest-with-langflow"
+                checked={disableIngestWithLangflow}
+                onCheckedChange={setDisableIngestWithLangflow}
+              />
+            </div>
             <div className="flex items-center justify-between py-3 border-b border-border">
               <div className="flex-1">
                 <Label

scripts/setup-e2e.sh

@@ -84,7 +84,7 @@ done
 
 echo "Waiting for Backend (via proxy)..."
 ELAPSED=0
-until curl -s http://localhost:8000/health >/dev/null; do
+until [ "$(curl -s http://localhost:8000/search/health -o /dev/null -w "%{http_code}")" -eq 200 ]; do
     sleep 5
     ELAPSED=$((ELAPSED + 5))
     if [ $ELAPSED -ge $TIMEOUT ]; then
@@ -95,4 +95,16 @@ until curl -s http://localhost:8000/health >/dev/null; do
     echo "Waiting for Backend... (${ELAPSED}s/${TIMEOUT}s)"
 done
 
+echo "Waiting for OpenSearch security configuration to be applied..."
+ELAPSED=0
+until ${CONTAINER_RUNTIME} logs os 2>&1 | grep -q "Security configuration applied successfully" || [ $ELAPSED -ge 60 ]; do
+    sleep 2
+    ELAPSED=$((ELAPSED + 2))
+done
+if [ $ELAPSED -ge 60 ]; then
+    echo "WARNING: OpenSearch security configuration wait timed out (60s)"
+else
+    echo "OpenSearch security configuration applied successfully"
+fi
+
 echo "Infrastructure Ready!"

src/api/health.py

@@ -17,7 +17,7 @@ async def health_check(request: Request):
     return JSONResponse({"status": "ok"}, status_code=200)
 
 
-async def opensearch_health_ready(request):
+async def opensearch_health_ready(request: Request):
     """Readiness probe: verifies OpenSearch dependency is reachable."""
     from config.settings import IBM_AUTH_ENABLED, OPENSEARCH_URL
 

src/api/router.py

@@ -1,13 +1,12 @@
 """Router endpoints that automatically route based on configuration settings."""
 
 import json
-import os
 import tempfile
 
 from fastapi import Depends, File, Form, UploadFile
 from fastapi.responses import JSONResponse
 
-from config.settings import DISABLE_INGEST_WITH_LANGFLOW
+from config.settings import get_openrag_config
 from dependencies import (
     get_current_user,
     get_document_service,
@@ -40,21 +39,22 @@ async def upload_ingest_router(
     - If DISABLE_INGEST_WITH_LANGFLOW is True: uses traditional OpenRAG upload
     - 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
     logger.debug(
         "Router upload_ingest endpoint called",
-        disable_langflow_ingest=DISABLE_INGEST_WITH_LANGFLOW,
+        disable_langflow_ingest=disable_ingest_with_langflow,
     )
 
-    if DISABLE_INGEST_WITH_LANGFLOW:
-        logger.debug("Routing to traditional OpenRAG upload")
-        # Route to traditional upload — just take the first file
-        from api.upload import upload as traditional_upload_fn
-
-        return await traditional_upload_fn(
-            file=file[0] if file else None,
-            document_service=document_service,
+    if disable_ingest_with_langflow:
+        logger.debug("Routing to traditional OpenRAG upload via task service")
+        return await _traditional_upload_ingest_task(
+            upload_files=file,
+            replace_duplicates=replace_duplicates.lower() == "true",
+            create_filter=create_filter.lower() == "true",
             session_manager=session_manager,
+            task_service=task_service,
             user=user,
+            settings_json=settings_json,
         )
 
     logger.debug("Routing to Langflow upload-ingest pipeline via task service")
@@ -72,6 +72,96 @@ async def upload_ingest_router(
     )
 
 
+async def _traditional_upload_ingest_task(
+    upload_files: list[UploadFile],
+    replace_duplicates: bool,
+    create_filter: bool,
+    session_manager,
+    task_service,
+    user: User,
+    settings_json: str | None = None,
+):
+    """Task-based traditional upload and ingest for single/multiple files"""
+    try:
+        if not upload_files:
+            return JSONResponse({"error": "Missing files"}, status_code=400)
+
+        settings = None
+        if settings_json:
+            try:
+                settings = json.loads(settings_json)
+            except json.JSONDecodeError as e:
+                return JSONResponse({"error": f"Invalid settings JSON: {e}"}, status_code=400)
+
+        user_id = user.user_id
+        user_name = user.name
+        user_email = user.email
+        jwt_token = user.jwt_token
+
+        temp_file_paths = []
+        original_filenames = []
+
+        try:
+            for upload_file in upload_files:
+                content = await upload_file.read()
+                original_filenames.append(upload_file.filename)
+                # Generate unique temp file to avoid collisions
+                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".tmp")
+                temp_path = temp_file.name
+                temp_file.close()
+                with open(temp_path, "wb") as f:
+                    f.write(content)
+                temp_file_paths.append(temp_path)
+
+            file_path_to_original_filename = dict(
+                zip(temp_file_paths, original_filenames, strict=True)
+            )
+
+            # Ensure the search index exists before creating the upload task
+            from api.documents import _ensure_index_exists
+
+            await _ensure_index_exists(jwt_token)
+
+            task_id = await task_service.create_upload_task(
+                user_id=user_id,
+                file_paths=temp_file_paths,
+                jwt_token=jwt_token,
+                owner_name=user_name,
+                owner_email=user_email,
+                original_filenames=file_path_to_original_filename,
+                replace_duplicates=replace_duplicates,
+                settings=settings,
+            )
+
+            return JSONResponse(
+                {
+                    "task_id": task_id,
+                    "message": f"Upload task created for {len(upload_files)} file(s)",
+                    "file_count": len(upload_files),
+                    "create_filter": create_filter,
+                    "filename": original_filenames[0] if len(original_filenames) == 1 else None,
+                },
+                status_code=202,
+            )
+
+        except Exception:
+            from utils.file_utils import safe_unlink
+
+            for temp_path in temp_file_paths:
+                safe_unlink(temp_path)
+            raise
+
+    except Exception as e:
+        logger.error("Task-based traditional upload_ingest failed", error=str(e))
+        import traceback
+
+        logger.error("Full traceback", traceback=traceback.format_exc())
+        error_msg = str(e)
+        if "AuthenticationException" in error_msg or "access denied" in error_msg.lower():
+            return JSONResponse({"error": "Access denied"}, status_code=403)
+        return JSONResponse({"error": "An internal error has occurred."}, status_code=500)
+
+
 async def _langflow_upload_ingest_task(
     upload_files: list[UploadFile],
     session_id,
@@ -113,13 +203,13 @@ async def _langflow_upload_ingest_task(
         original_filenames = []
 
         try:
-            temp_dir = tempfile.gettempdir()
-
             for upload_file in upload_files:
                 content = await upload_file.read()
                 original_filenames.append(upload_file.filename)
-                safe_filename = upload_file.filename.replace(" ", "_").replace("/", "_")
-                temp_path = os.path.join(temp_dir, safe_filename)
+                # Generate unique temp file to avoid collisions
+                temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".tmp")
+                temp_path = temp_file.name
+                temp_file.close()
                 with open(temp_path, "wb") as f:
                     f.write(content)
                 temp_file_paths.append(temp_path)

src/api/settings/endpoints.py

@@ -211,6 +211,7 @@ async def get_settings(
                 ocr=knowledge_config.ocr,
                 picture_descriptions=knowledge_config.picture_descriptions,
                 index_name=knowledge_config.index_name,
+                disable_ingest_with_langflow=knowledge_config.disable_ingest_with_langflow,
             ),
             agent=AgentConfig(
                 llm_model=agent_config.llm_model,
@@ -461,6 +462,15 @@ async def update_settings(
             except Exception as e:
                 logger.error(f"Failed to update docling settings in flow: {str(e)}")
 
+        if body.disable_ingest_with_langflow is not None:
+            current_config.knowledge.disable_ingest_with_langflow = (
+                body.disable_ingest_with_langflow
+            )
+            config_updated = True
+            logger.info(
+                f"Disable Langflow ingestion changed to {body.disable_ingest_with_langflow}"
+            )
+
         if body.chunk_size is not None:
             effective_overlap = (
                 body.chunk_overlap

src/api/settings/models.py

@@ -21,6 +21,7 @@ class SettingsUpdateBody(BaseModel):
     table_structure: bool | None = None
     ocr: bool | None = None
     picture_descriptions: bool | None = None
+    disable_ingest_with_langflow: bool | None = None
     embedding_model: str | None = Field(None, min_length=1)
     embedding_provider: str | None = Field(None, pattern="^(openai|watsonx|ollama)$")
     index_name: str | None = Field(None, min_length=1)
@@ -128,6 +129,7 @@ class KnowledgeConfig(BaseModel):
     ocr: bool | None
     picture_descriptions: bool | None
     index_name: str | None
+    disable_ingest_with_langflow: bool | None
 
 
 class AgentConfig(BaseModel):