Merge branch 'release-1.12.0' of https://github.com/langflow-ai/langflow into sso-admin-settings

Author: deon-sanchez

.secrets.baseline

@@ -103,12 +103,6 @@
       "path": "/settings/messages",
       "surface": "Messages settings page",
       "ready": [{ "testId": "settings_menu_header" }]
-    },
-    {
-      "id": "account-delete",
-      "path": "/account/delete",
-      "surface": "Delete account page",
-      "ready": [{ "role": "button", "name": "delete", "first": true }]
     }
   ],
   "dynamic": [

scripts/a11y/a11y_routes.json

@@ -12,7 +12,7 @@
 from fastapi import HTTPException
 from lfx.cli.script_loader import extract_structured_result
 from lfx.events.event_manager import EventManager, create_default_event_manager
-from lfx.execution import get_default_coordinator
+from lfx.execution import aget_default_coordinator
 from lfx.log.logger import logger
 from lfx.schema.schema import InputValueRequest
 from lfx.utils.flow_validation import CustomComponentValidationError
@@ -60,11 +60,9 @@ async def _run_graph_with_events(
         graph.prepare()
         inputs = InputValueRequest(input_value=input_value) if input_value else None
 
+        coordinator = await aget_default_coordinator()
         results = [
-            payload
-            async for payload in get_default_coordinator().stream(
-                graph, initial_inputs=inputs, event_manager=event_manager
-            )
+            payload async for payload in coordinator.stream(graph, initial_inputs=inputs, event_manager=event_manager)
         ]
         execution_result.result = extract_structured_result(results)
     except Exception as e:  # noqa: BLE001
@@ -137,7 +135,8 @@ async def execute_flow_file(
         graph.prepare()
         inputs = InputValueRequest(input_value=input_value) if input_value else None
 
-        results = [payload async for payload in get_default_coordinator().stream(graph, initial_inputs=inputs)]
+        coordinator = await aget_default_coordinator()
+        results = [payload async for payload in coordinator.stream(graph, initial_inputs=inputs)]
         flow_result = extract_structured_result(results)
     except HTTPException:
         raise

src/backend/base/langflow/agentic/services/flow_executor.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from contextlib import asynccontextmanager
 from typing import TYPE_CHECKING, Union
 
@@ -248,10 +249,45 @@ def get_queue_service() -> JobQueueService:
     return get_service(ServiceType.JOB_QUEUE_SERVICE, JobQueueServiceFactory())
 
 
+def _is_lfx_no_op_auth_service(service: object) -> bool:
+    """Report whether ``service`` is LFX's no-op auth stub.
+
+    Resolved through ``sys.modules`` rather than a direct import so that merely asking the
+    question cannot import ``lfx.services.auth.service`` -- that import is what registers
+    the stub for the ``auth_service`` slot in the first place. If the module was never
+    imported the stub cannot be the resolved service anyway.
+    """
+    lfx_auth_module = sys.modules.get("lfx.services.auth.service")
+    return lfx_auth_module is not None and isinstance(service, lfx_auth_module.AuthService)
+
+
 def get_auth_service() -> BaseAuthService:
-    """Retrieve the authentication service."""
+    """Retrieve the authentication service.
+
+    LFX's no-op ``AuthService`` claims the ``auth_service`` slot at import time via
+    ``@register_service(..., override=True)``, and a registered service class always wins
+    over the factory passed here as a fallback. That stub must never be the resolved
+    service in a Langflow process: it raises ``NotImplementedError`` for user/token
+    operations and answers API-key checks with ``None``. ``register_all_service_factories()``
+    overrides it during startup, so anything touching auth before (or without) that call
+    would otherwise get the stub -- swap it out for Langflow's implementation here.
+
+    A plugin-provided auth service (e.g. SSO, registered through ``lfx.toml``) is a
+    different class and is left untouched.
+    """
     from langflow.services.auth.factory import AuthServiceFactory
 
+    service = get_service(ServiceType.AUTH_SERVICE, AuthServiceFactory())
+    if not _is_lfx_no_op_auth_service(service):
+        return service
+
+    from lfx.services.manager import get_service_manager
+
+    from langflow.services.auth.service import AuthService
+
+    service_manager = get_service_manager()
+    service_manager.register_service_class(ServiceType.AUTH_SERVICE, AuthService, override=True)
+    service_manager.update(ServiceType.AUTH_SERVICE)
     return get_service(ServiceType.AUTH_SERVICE, AuthServiceFactory())
 
 

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

@@ -17,7 +17,7 @@ def cleanup_test_files(self):
         """Clean up test files after all tests in the class complete."""
         yield
         # Clean up test files created during tests
-        test_files = ["test_data.json", "test_message.txt", "test_output.csv", "test_page.html"]
+        test_files = ["test_data.json", "test_message.txt", "test_output.csv", "test_page.html", "test_s3_output.txt"]
         for filename in test_files:
             filepath = Path(filename)
             if filepath.exists():
@@ -109,7 +109,7 @@ async def test_save_dataframe_to_csv(self, component_class):
             mock_db = AsyncMock()
             mock_session.return_value.__aenter__.return_value = mock_db
             mock_get_user.return_value = MagicMock()
-            mock_upload.return_value = "test_output.csv"
+            mock_upload.return_value = MagicMock(path="test_output.csv", provider="s3")
 
             # Execute - real temp file creation, real DataFrame.to_csv(), real cleanup
             result = await component.save_to_file()
@@ -143,7 +143,7 @@ async def test_save_data_to_json(self, component_class):
             mock_db = AsyncMock()
             mock_session.return_value.__aenter__.return_value = mock_db
             mock_get_user.return_value = MagicMock()
-            mock_upload.return_value = "test_data.json"
+            mock_upload.return_value = MagicMock(path="test_data.json", provider="s3")
 
             result = await component.save_to_file()
 
@@ -175,13 +175,172 @@ async def test_save_message_to_txt(self, component_class):
             mock_db = AsyncMock()
             mock_session.return_value.__aenter__.return_value = mock_db
             mock_get_user.return_value = MagicMock()
-            mock_upload.return_value = "test_message.txt"
+            mock_upload.return_value = MagicMock(path="test_message.txt", provider="s3")
 
             result = await component.save_to_file()
 
             assert "saved successfully" in result.text
             assert "test_message.txt" in result.text
 
+    @pytest.mark.asyncio
+    async def test_save_local_mode_with_s3_backend_cleans_staging_and_reports_storage_path(
+        self, component_class, tmp_path
+    ):
+        """Remote (S3) backend: Local mode deletes the staging file and reports the storage path.
+
+        This guards the fix for the leak where Local mode left a redundant copy in
+        cwd and surfaced a misleading local path when the backend was S3.
+        """
+        component = component_class(_user_id=str(uuid4()))
+        message = Message(text="This should end up only in S3")
+        component.set_attributes(
+            {
+                "input": message,
+                "file_name": "test_s3_output",
+                "local_format": "txt",
+                "storage_location": [{"name": "Local"}],
+            }
+        )
+
+        # upload_user_file returns the durable storage location + provider
+        upload_response = MagicMock()
+        upload_response.path = "files/user-uuid/test_s3_output.txt"
+        upload_response.provider = "s3"
+
+        # Force the storage backend to look remote (S3) without restricting paths
+        settings_mock = MagicMock()
+        settings_mock.storage_type = "s3"
+        settings_mock.restrict_local_file_access = False
+        settings_mock.config_dir = str(tmp_path)
+        settings_service_mock = MagicMock()
+        settings_service_mock.settings = settings_mock
+
+        with (
+            patch("langflow.api.v2.files.upload_user_file", new_callable=AsyncMock) as mock_upload,
+            patch("lfx.services.deps.session_scope") as mock_session,
+            patch(
+                "langflow.services.database.models.user.crud.get_user_by_id", new_callable=AsyncMock
+            ) as mock_get_user,
+            patch(
+                "lfx.components.files_and_knowledge.save_file.get_settings_service",
+                return_value=settings_service_mock,
+            ),
+        ):
+            mock_db = AsyncMock()
+            mock_session.return_value.__aenter__.return_value = mock_db
+            mock_get_user.return_value = MagicMock()
+            mock_upload.return_value = upload_response
+
+            result = await component.save_to_file()
+
+        # Message reports the durable storage destination + provider, not a local path
+        assert "files/user-uuid/test_s3_output.txt" in result.text
+        assert "S3" in result.text
+        assert str(tmp_path) not in result.text
+        # The local staging file was cleaned up
+        assert not (Path.cwd() / "test_s3_output.txt").exists()
+
+    @pytest.mark.asyncio
+    async def test_append_mode_remote_backend_accumulates_across_calls(self, component_class, tmp_path):
+        """append_mode + remote (S3) backend keeps accumulating, not resetting to overwrite.
+
+        Guards against the staging cleanup deleting the local accumulator that append
+        relies on (should_append hinges on the file persisting across calls). Without
+        the append exception, the second call would silently overwrite.
+        """
+        file_name = "test_append_remote"
+        staging = Path.cwd() / f"{file_name}.txt"
+        if staging.exists():
+            staging.unlink()
+
+        settings_mock = MagicMock()
+        settings_mock.storage_type = "s3"
+        settings_mock.restrict_local_file_access = False
+        settings_mock.config_dir = str(tmp_path)
+        settings_service_mock = MagicMock()
+        settings_service_mock.settings = settings_mock
+
+        upload_response = MagicMock()
+        upload_response.path = f"files/uid/{file_name}.txt"
+        upload_response.provider = "s3"
+
+        def make_component(text):
+            component = component_class(_user_id=str(uuid4()))
+            component.set_attributes(
+                {
+                    "input": Message(text=text),
+                    "file_name": file_name,
+                    "local_format": "txt",
+                    "append_mode": True,
+                    "storage_location": [{"name": "Local"}],
+                }
+            )
+            return component
+
+        try:
+            with (
+                patch("langflow.api.v2.files.upload_user_file", new_callable=AsyncMock) as mock_upload,
+                patch("lfx.services.deps.session_scope") as mock_session,
+                patch(
+                    "langflow.services.database.models.user.crud.get_user_by_id", new_callable=AsyncMock
+                ) as mock_get_user,
+                patch(
+                    "lfx.components.files_and_knowledge.save_file.get_settings_service",
+                    return_value=settings_service_mock,
+                ),
+            ):
+                mock_db = AsyncMock()
+                mock_session.return_value.__aenter__.return_value = mock_db
+                mock_get_user.return_value = MagicMock()
+                mock_upload.return_value = upload_response
+
+                await make_component("line one").save_to_file()
+                # Staging file must survive so the next call can append to it
+                assert staging.exists()
+                await make_component("line two").save_to_file()
+
+            # Content accumulated across both calls — not overwritten
+            assert staging.read_text(encoding="utf-8") == "line one\nline two"
+        finally:
+            if staging.exists():
+                staging.unlink()
+
+    @pytest.mark.asyncio
+    async def test_save_aws_mode_namespaces_key_by_user_id(self, component_class):
+        """AWS mode namespaces the S3 key by user_id so multi-user runs don't collide.
+
+        Layout must be {s3_prefix}/{user_id}/{file_name}.{ext} — otherwise every
+        user writing the same file_name overwrites the same key.
+        """
+        user_id = str(uuid4())
+        component = component_class(_user_id=user_id)
+        component.set_attributes(
+            {
+                "input": Message(text="hello aws"),
+                "file_name": "report",
+                "aws_format": "txt",
+                "storage_location": [{"name": "AWS"}],
+                "aws_access_key_id": "test-access-key",  # pragma: allowlist secret
+                "aws_secret_access_key": "test-secret-key",  # pragma: allowlist secret
+                "bucket_name": "my-bucket",
+                "aws_region": "us-east-1",
+                "s3_prefix": "files",
+            }
+        )
+
+        mock_s3 = MagicMock()
+        with (
+            patch("boto3.client", return_value=mock_s3),
+            patch("lfx.base.data.cloud_storage_utils.validate_aws_credentials"),
+        ):
+            result = await component.save_to_file()
+
+        # upload_file(temp_path, bucket, key) — third positional arg is the S3 key
+        mock_s3.upload_file.assert_called_once()
+        key = mock_s3.upload_file.call_args[0][2]
+        assert key == f"files/{user_id}/report.txt"
+        assert "my-bucket" in result.text
+
     @pytest.mark.asyncio
     async def test_save_message_to_html(self, component_class):
         """Test saving Message to html format."""
@@ -207,7 +366,7 @@ async def test_save_message_to_html(self, component_class):
             mock_db = AsyncMock()
             mock_session.return_value.__aenter__.return_value = mock_db
             mock_get_user.return_value = MagicMock()
-            mock_upload.return_value = "test_page.html"
+            mock_upload.return_value = MagicMock(path="test_page.html", provider="s3")
 
             result = await component.save_to_file()
 
@@ -319,7 +478,7 @@ async def test_file_name_with_extension_stripped(self, component_class):
             mock_db = AsyncMock()
             mock_session.return_value.__aenter__.return_value = mock_db
             mock_get_user.return_value = MagicMock()
-            mock_upload.return_value = "test_output.csv"
+            mock_upload.return_value = MagicMock(path="test_output.csv", provider="s3")
 
             result = await component.save_to_file()
 
@@ -349,6 +508,16 @@ async def test_append_mode_txt_file(self, component_class):
                 }
             )
 
+            # This test verifies LOCAL-backend append semantics (the file persists on
+            # disk and is re-read), so pin the backend to local. Under a remote (S3)
+            # backend the staging file is intentionally deleted after upload.
+            settings_mock = MagicMock()
+            settings_mock.storage_type = "local"
+            settings_mock.restrict_local_file_access = False
+            settings_mock.config_dir = str(tmp_path.parent)
+            settings_service_mock = MagicMock()
+            settings_service_mock.settings = settings_mock
+
             # Mock the path resolution to return our temp file
             with (
                 patch("lfx.components.files_and_knowledge.save_file.Path") as mock_path_class,
@@ -357,13 +526,17 @@ async def test_append_mode_txt_file(self, component_class):
                 patch(
                     "langflow.services.database.models.user.crud.get_user_by_id", new_callable=AsyncMock
                 ) as mock_get_user,
+                patch(
+                    "lfx.components.files_and_knowledge.save_file.get_settings_service",
+                    return_value=settings_service_mock,
+                ),
             ):
                 # Make Path() return our temp file path
                 mock_path_class.return_value = tmp_path
                 mock_db = AsyncMock()
                 mock_session.return_value.__aenter__.return_value = mock_db
                 mock_get_user.return_value = MagicMock()
-                mock_upload.return_value = tmp_path.name
+                mock_upload.return_value = MagicMock(path=tmp_path.name, provider="local")
 
                 result = await component.save_to_file()
 

src/backend/base/langflow/services/deps.py

@@ -86,3 +86,33 @@ async def test_get_current_user_delegates_to_service(dummy_auth_registration):
     assert ("get_current_user", None, "q", None, None) in dummy.calls
     assert response["user"] == "dummy"
     assert response["db"] is db
+
+
+def test_get_auth_service_replaces_lfx_no_op_stub():
+    """LFX's no-op auth service must never be what Langflow resolves.
+
+    It self-registers for the ``auth_service`` slot at import time, so any process that
+    touches auth before ``register_all_service_factories()`` would otherwise get a service
+    that raises ``NotImplementedError`` for user/token operations and answers API-key
+    checks with ``None``.
+    """
+    from langflow.services.auth.service import AuthService as LangflowAuthService
+    from langflow.services.deps import get_auth_service
+    from lfx.services.auth.service import AuthService as LFXNoOpAuthService
+
+    service_manager = get_service_manager()
+    previous_class = service_manager.service_classes.get(ServiceType.AUTH_SERVICE)
+    previous_instance = service_manager.services.pop(ServiceType.AUTH_SERVICE, None)
+    service_manager.register_service_class(ServiceType.AUTH_SERVICE, LFXNoOpAuthService, override=True)
+
+    try:
+        assert isinstance(service_manager.get(ServiceType.AUTH_SERVICE), LFXNoOpAuthService)
+        assert isinstance(get_auth_service(), LangflowAuthService)
+    finally:
+        service_manager.services.pop(ServiceType.AUTH_SERVICE, None)
+        if previous_class is not None:
+            service_manager.service_classes[ServiceType.AUTH_SERVICE] = previous_class
+        else:
+            service_manager.service_classes.pop(ServiceType.AUTH_SERVICE, None)
+        if previous_instance is not None:
+            service_manager.services[ServiceType.AUTH_SERVICE] = previous_instance