Merge remote-tracking branch 'origin/release-1.11.3' into fix/le-2134-chatinput-files-1113-resolve

# Conflicts:
#	src/backend/base/langflow/api/build.py
#	src/backend/tests/unit/api/test_build_source_flow_provenance.py
#	src/lfx/src/lfx/_assets/component_index.json
#	src/lfx/src/lfx/graph/graph/base.py

Author: erichare

src/backend/base/langflow/api/build.py

@@ -13,6 +13,7 @@
 from lfx.log.logger import logger
 from lfx.schema.legacy_render import project_payload_to_v1
 from lfx.schema.schema import InputValueRequest
+from lfx.utils.file_path_security import LocalFileAccessError
 from sqlmodel import select
 
 from langflow.api.disconnect import DisconnectHandlerStreamingResponse
@@ -571,7 +572,7 @@ async def build_graph_and_get_order() -> tuple[list[str], list[str], Graph]:
                 error_message=str(exc),
             )
 
-            if "stream or streaming set to True" in str(exc):
+            if isinstance(exc, LocalFileAccessError) or "stream or streaming set to True" in str(exc):
                 raise HTTPException(status_code=400, detail=str(exc)) from exc
             await logger.aexception("Error checking build status: " + str(exc))
             raise HTTPException(status_code=500, detail=str(exc)) from exc
@@ -616,11 +617,17 @@ async def create_graph(fresh_session, flow_id_str: str, flow_name: str | None) -
             )
         else:
             if not flow_name:
-                result = await fresh_session.exec(select(Flow.name).where(Flow.id == flow_id))
+                lookup_flow_id = source_flow_id if source_flow_id is not None else flow_id
+                result = await fresh_session.exec(select(Flow.name).where(Flow.id == lookup_flow_id))
                 flow_name = result.first()
 
+            # Sanitized public data still contains FileInput references under
+            # the real source-flow namespace. Build under that trusted scope;
+            # after parameter containment completes, switch the graph to the
+            # visitor-virtual execution ID below.
+            graph_build_flow_id = str(source_flow_id) if source_flow_id is not None else flow_id_str
             graph = await build_graph_from_data(
-                flow_id=flow_id_str,
+                flow_id=graph_build_flow_id,
                 payload=data.model_dump(),
                 user_id=str(current_user.id),
                 flow_name=flow_name,
@@ -630,8 +637,8 @@ async def create_graph(fresh_session, flow_id_str: str, flow_name: str | None) -
 
         if source_flow_id is not None:
             # This value comes from the server-resolved public flow, never from
-            # request data. ChatInput uses it only to validate already-approved
-            # public attachment references after flow_id becomes visitor-virtual.
+            # request data. FileInput containment and ChatInput attachment
+            # validation use it only after flow_id becomes visitor-virtual.
             graph.source_flow_id = str(source_flow_id)
             graph.flow_id = str(flow_id)
         return graph

src/backend/base/langflow/api/v1/chat.py

@@ -15,6 +15,7 @@
 from lfx.services.cache.utils import CacheMiss
 from lfx.utils.flow_validation import (
     CustomComponentValidationError,
+    prepare_flow_build_for_user,
     prepare_public_flow_build,
     validate_flow_for_current_settings,
     validate_public_flow_no_code_execution,
@@ -177,6 +178,12 @@ async def retrieve_vertices_order(
         if not data:
             graph = await build_graph_from_db(flow_id=flow_id, session=session, chat_service=chat_service)
         else:
+            sanitized_data = await prepare_flow_build_for_user(
+                data.model_dump(),
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                data = FlowDataRequest.model_validate(sanitized_data)
             graph = await build_and_cache_graph_from_data(
                 flow_id=flow_id, graph_data=data.model_dump(), chat_service=chat_service
             )
@@ -214,6 +221,8 @@ async def retrieve_vertices_order(
             raise HTTPException(status_code=400, detail=str(exc)) from exc
         if isinstance(exc, CustomComponentValidationError):
             raise HTTPException(status_code=400, detail=str(exc)) from exc
+        if isinstance(exc, RuntimeError):
+            raise HTTPException(status_code=503, detail=str(exc)) from exc
         await logger.aexception("Error checking build status")
         raise HTTPException(status_code=500, detail=str(exc)) from exc
 
@@ -308,7 +317,13 @@ async def build_flow(
 
     try:
         if data:
-            validate_flow_for_current_settings(data.model_dump())
+            raw_data = data.model_dump()
+            sanitized_data = await prepare_flow_build_for_user(
+                raw_data,
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                data = FlowDataRequest.model_validate(sanitized_data)
         elif flow and flow.data:
             validate_flow_for_current_settings(flow.data)
     except CustomComponentValidationError as exc:

src/backend/base/langflow/api/v1/flows.py

@@ -8,10 +8,11 @@
 from uuid import UUID
 
 import orjson
-from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile
+from fastapi import APIRouter, Depends, File, HTTPException, Request, UploadFile, status
 from fastapi.encoders import jsonable_encoder
 from fastapi_pagination import Page, Params
 from fastapi_pagination.ext.sqlmodel import apaginate
+from lfx.log.logger import logger
 from lfx.services.cache.utils import CACHE_MISS
 from pydantic import ValidationError
 from sqlmodel import and_, col, select
@@ -56,6 +57,10 @@
 from langflow.services.authorization.fetch import deny_to_404
 from langflow.services.authorization.utils import _resolve_authz_domain
 from langflow.services.cache.service import ThreadingInMemoryCache
+from langflow.services.database.lock_retry import (
+    is_database_lock_error,
+    run_with_lock_retry,
+)
 from langflow.services.database.models.deployment.exceptions import (
     araise_if_deployment_guard_error_or_skip,
 )
@@ -74,6 +79,7 @@
 # and FlowVersionError from the flow_version modules.
 from langflow.services.database.models.folder.constants import DEFAULT_FOLDER_NAME
 from langflow.services.database.models.folder.model import Folder
+from langflow.services.database.models.user.model import UserRead
 from langflow.services.deps import get_settings_service, get_storage_service
 from langflow.services.storage.service import StorageService
 from langflow.utils.compression import compress_response
@@ -103,6 +109,9 @@ def _handle_unique_constraint_error(exc: Exception, *, status_code: int = 400) -
 # build router
 router = APIRouter(prefix="/flows", tags=["Flows"])
 
+FLOW_UPDATE_FAILED = "Could not update the flow."
+FLOW_UPDATE_BUSY = "The database is busy. Please retry the request."
+
 
 @router.post("/", response_model=FlowRead, status_code=201)
 async def create_flow(
@@ -328,6 +337,7 @@ async def update_flow(
     storage_service: Annotated[StorageService, Depends(get_storage_service)],
 ):
     """Update a flow."""
+    actor = UserRead.model_validate(current_user, from_attributes=True)
     try:
         # Destination check: if the payload moves the flow into a new
         # workspace/folder, the caller must also be authorized to write at the
@@ -339,7 +349,7 @@ async def update_flow(
         if target_workspace_id != db_flow.workspace_id or target_folder_id != db_flow.folder_id:
             try:
                 await ensure_flow_permission(
-                    current_user,
+                    actor,
                     FlowAction.WRITE,
                     flow_id=flow_id,
                     flow_user_id=db_flow.user_id,
@@ -357,7 +367,7 @@ async def update_flow(
 
         async def operation() -> FlowRead:
             # Re-load inside each attempt so retry after nested rollback never uses an expired ORM instance.
-            db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=current_user.id)
+            db_flow_for_attempt = await _read_flow(session=session, flow_id=flow_id, user_id=actor.id)
             if not db_flow_for_attempt:
                 raise HTTPException(status_code=404, detail="Flow not found")
             # TOCTOU: a concurrent PATCH could have moved this flow to a
@@ -367,7 +377,7 @@ async def operation() -> FlowRead:
             # stale check across a race.
             try:
                 await ensure_flow_permission(
-                    current_user,
+                    actor,
                     FlowAction.WRITE,
                     flow_id=flow_id,
                     flow_user_id=db_flow_for_attempt.user_id,
@@ -386,7 +396,7 @@ async def operation() -> FlowRead:
             ):
                 try:
                     await ensure_flow_permission(
-                        current_user,
+                        actor,
                         FlowAction.WRITE,
                         flow_id=flow_id,
                         flow_user_id=db_flow_for_attempt.user_id,
@@ -399,26 +409,47 @@ async def operation() -> FlowRead:
                 session=session,
                 db_flow=db_flow_for_attempt,
                 flow=flow,
-                user_id=current_user.id,
+                user_id=actor.id,
                 storage_service=storage_service,
             )
 
-        if folder_id_will_change:
-            return await retry_flow_operation_on_deployment_guard(
-                db=session,
-                user_id=current_user.id,
-                flow_ids=[flow_id],
-                operation=operation,
-            )
-        return await operation()
+        async def update_attempt(_attempt: int) -> FlowRead:
+            if folder_id_will_change:
+                return await retry_flow_operation_on_deployment_guard(
+                    db=session,
+                    user_id=actor.id,
+                    flow_ids=[flow_id],
+                    operation=operation,
+                )
+            return await operation()
+
+        return await run_with_lock_retry(
+            update_attempt,
+            session=session,
+            description=f"update_flow {flow_id}",
+        )
     except HTTPException:
         raise
     except Exception as e:
         await araise_if_deployment_guard_error_or_skip(
             e,
             log_message=f"op=update_flow flow_id={flow_id}",
         )
-        raise _handle_unique_constraint_error(e) from e
+        if is_database_lock_error(e):
+            await logger.awarning("op=update_flow flow_id=%s exhausted lock retries", flow_id)
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail=FLOW_UPDATE_BUSY,
+                headers={"Retry-After": "1"},
+            ) from e
+        handled_error = _handle_unique_constraint_error(e)
+        if handled_error.status_code != status.HTTP_500_INTERNAL_SERVER_ERROR:
+            raise handled_error from e
+        await logger.aerror("op=update_flow flow_id=%s failed with %s", flow_id, type(e).__name__)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=FLOW_UPDATE_FAILED,
+        ) from e
 
 
 @router.put("/{flow_id}", response_model=FlowRead)

src/backend/base/langflow/api/v2/workflow.py

@@ -221,8 +221,8 @@ async def authorize_flow_action(
         ) from err
 
 
-def _apply_execution_gates(parsed, flow, current_user: UserRead) -> None:
-    """The langflow request gates that run before a flow executes."""
+def _apply_execution_gates(parsed, flow, current_user: UserRead):
+    """Run request gates and return any server-sanitized execution payload."""
     _reject_unsupported_sync_fields(parsed)
     _reject_sync_only_fields(parsed)
     try:
@@ -234,7 +234,7 @@ def _apply_execution_gates(parsed, flow, current_user: UserRead) -> None:
         if exc.status_code == status.HTTP_404_NOT_FOUND:
             raise _flow_not_found_http_exception(str(parsed.flow_id)) from exc
         raise
-    _validate_flow_data_for_execution(parsed, flow)
+    return _validate_flow_data_for_execution(parsed, flow, current_user)
 
 
 async def run_sync_with_mapping(
@@ -246,7 +246,7 @@ async def run_sync_with_mapping(
     background_tasks: BackgroundTasks,
 ) -> WorkflowExecutionResponse:
     """Inline sync run with the langflow timeout/validation error mapping."""
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     job_id = uuid4()
     try:
         return await execute_sync_workflow_with_timeout(
@@ -321,7 +321,7 @@ def build_stream_response(
     side-channel, and vertex-build persistence all survive. Validation gates run
     before the response is constructed so a bad request fails before streaming.
     """
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     adapter = get_stream_adapter(
         stream_protocol,
         StreamAdapterContext(
@@ -346,7 +346,7 @@ async def submit_background_with_mapping(
     stream_protocol: str,
 ) -> WorkflowJobResponse:
     """Queue a durable background run with the langflow service error mapping."""
-    _apply_execution_gates(parsed, flow, current_user)
+    parsed = _apply_execution_gates(parsed, flow, current_user)
     try:
         return await execute_workflow_background(
             parsed=parsed,

src/backend/base/langflow/api/v2/workflow_validation.py

@@ -8,8 +8,14 @@
 
 from __future__ import annotations
 
+from dataclasses import replace
+
 from fastapi import HTTPException, status
-from lfx.utils.flow_validation import CustomComponentValidationError, validate_flow_for_current_settings
+from lfx.utils.flow_validation import (
+    CustomComponentValidationError,
+    prepare_flow_build_for_user_from_cache,
+    validate_flow_for_current_settings,
+)
 from lfx.workflow.converters import ParsedWorkflowRun
 
 from langflow.services.authorization.fetch import deny_to_404
@@ -85,17 +91,27 @@ def _enforce_flow_data_override_owner(parsed: ParsedWorkflowRun, flow: FlowRead,
     )
 
 
-def _validate_flow_data_for_execution(parsed: ParsedWorkflowRun, flow: FlowRead) -> None:
-    """Apply the same server-side component policy gate used by v1/public runs."""
+def _validate_flow_data_for_execution(
+    parsed: ParsedWorkflowRun,
+    flow: FlowRead,
+    current_user: UserRead,
+) -> ParsedWorkflowRun:
+    """Apply component policies and return sanitized caller-supplied graph data."""
     try:
         if parsed.data is not None:
-            validate_flow_for_current_settings(parsed.data)
+            sanitized_data = prepare_flow_build_for_user_from_cache(
+                parsed.data,
+                is_superuser=current_user.is_superuser,
+            )
+            if sanitized_data is not None:
+                return replace(parsed, data=sanitized_data)
         elif flow.data:
             validate_flow_for_current_settings(flow.data)
     except CustomComponentValidationError as exc:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
     except RuntimeError as exc:
         raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=str(exc)) from exc
+    return parsed
 
 
 def _validate_output_ids(output_ids: list[str] | None, terminal_node_ids: list[str]) -> None:

src/backend/base/langflow/initial_setup/starter_projects/Content Aggregator.json

@@ -16,10 +16,11 @@ class of failure — the only valid recovery is to end the transaction and run i
 
 import asyncio
 import random
+import sqlite3
 from typing import TYPE_CHECKING, TypeVar
 
 from lfx.log.logger import logger
-from sqlalchemy.exc import DBAPIError, SQLAlchemyError
+from sqlalchemy.exc import SQLAlchemyError
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -55,7 +56,10 @@ def is_database_lock_error(exc: BaseException | None) -> bool:
         error_name = getattr(exc, "sqlite_errorname", None)
         if error_name in _SQLITE_LOCK_ERROR_NAMES:
             return True
-        if isinstance(exc, DBAPIError | SQLAlchemyError) or error_name is not None:
+        # SQLAlchemy's wrapper text contains the SQL statement and bound
+        # parameters. Inspect only the underlying SQLite operational error so
+        # user data containing "database is locked" cannot trigger retries.
+        if isinstance(exc, sqlite3.OperationalError):
             message = str(exc).lower()
             if any(marker in message for marker in _SQLITE_LOCK_MESSAGES):
                 return True