fix: enforce locked flows in backend

src/backend/base/langflow/agentic/utils/assistant_runner.py

@@ -23,6 +23,7 @@
 from langflow.agentic.services.flow_types import LANGFLOW_ASSISTANT_FLOW
 from langflow.api.v1.flows import _new_flow, _save_flow_to_fs
 from langflow.initial_setup.setup import get_or_create_default_folder
+from langflow.services.database.models.flow.guards import ensure_flow_unlocked, lock_flow_for_update
 from langflow.services.database.models.flow.model import Flow, FlowCreate
 from langflow.services.deps import get_storage_service
 
@@ -179,6 +180,10 @@ async def run_assistant_and_persist(
     ``model_name``.
     """
     flow, created_new = await _ensure_flow(session, user_id, flow_id)
+    if not created_new:
+        # Fail before invoking the model. Locked flows are read-only to the
+        # headless assistant just as they are to direct MCP edit tools.
+        ensure_flow_unlocked(flow)
 
     request = AssistantRequest(
         flow_id=str(flow.id),
@@ -205,6 +210,11 @@ async def run_assistant_and_persist(
     canvas, working_snapshot, result_text, error_text, field_edits = await _consume_stream(stream, flow.data)
 
     if canvas.changed:
+        if not created_new:
+            # The assistant can run for a while. Re-read under a row lock so a
+            # concurrent lock toggle and this write are ordered atomically.
+            await lock_flow_for_update(session, flow)
+            ensure_flow_unlocked(flow)
         flow_data = working_snapshot["data"] if working_snapshot else canvas.data
         # Headless MCP has no UI to apply an edit_field review proposal, so apply
         # each to the working flow here or the text edit is dropped (Bug #13641).

src/backend/base/langflow/agentic/utils/flow_component.py

@@ -7,6 +7,7 @@
 from lfx.log.logger import logger
 
 from langflow.helpers.flow import get_flow_by_id_or_endpoint_name
+from langflow.services.database.models.flow.guards import ensure_flow_unlocked, lock_flow_for_update
 from langflow.services.database.models.flow.model import Flow
 from langflow.services.deps import session_scope
 
@@ -231,6 +232,8 @@ async def update_component_field_value(
         if flow.data is None:
             return {"error": f"Flow {flow_id_or_name} has no data", "success": False}
 
+        ensure_flow_unlocked(flow)
+
         flow_id_str = str(flow.id)
 
         # Find the component in the flow data
@@ -271,10 +274,15 @@ async def update_component_field_value(
             if not db_flow:
                 return {"error": f"Flow {flow_id_str} not found in database", "success": False}
 
+            await lock_flow_for_update(session, db_flow)
+
             # Verify user has permission
             if str(db_flow.user_id) != str(user_id):
                 return {"error": "User does not have permission to update this flow", "success": False}
 
+            # Check the row while its write lock is held through commit.
+            ensure_flow_unlocked(db_flow)
+
             # Update the flow data
             db_flow.data = flow_data
             session.add(db_flow)

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

@@ -24,6 +24,11 @@
 from langflow.api.utils import build_content_disposition, normalize_flow_for_export, remove_api_keys
 from langflow.services.database.models.base import orjson_dumps
 from langflow.services.database.models.deployment.orm_guards import ensure_flow_move_allowed
+from langflow.services.database.models.flow.guards import (
+    LockedFlowError,
+    ensure_flow_update_allowed,
+    lock_flow_for_update,
+)
 from langflow.services.database.models.flow.model import (
     Flow,
     FlowCreate,
@@ -153,6 +158,14 @@ def _endpoint_name_was_explicitly_cleared(flow: FlowCreate | FlowUpdate) -> bool
     return "endpoint_name" in flow.model_fields_set and flow.endpoint_name in (None, "")
 
 
+def _ensure_api_flow_update_allowed(db_flow: Flow, update_data: dict[str, Any]) -> None:
+    """Translate the domain lock guard into the API's 423 response."""
+    try:
+        ensure_flow_update_allowed(db_flow, update_data)
+    except LockedFlowError as exc:
+        raise HTTPException(status_code=423, detail=str(exc)) from exc
+
+
 async def _verify_fs_path(path: str | None, user_id: UUID, storage_service: StorageService) -> None:
     """Verify and prepare the filesystem path for flow storage."""
     if path is not None:
@@ -387,6 +400,8 @@ async def _update_existing_flow(
     actor. This mirrors the cross-user semantics already enforced by
     ``_patch_flow``.
     """
+    await lock_flow_for_update(session, existing_flow)
+
     settings_service = get_settings_service()
     actor_user_id = current_user.id
     owner_user_id: UUID = existing_flow.user_id
@@ -494,6 +509,8 @@ async def _update_existing_flow(
             new_folder_id=update_data["folder_id"],
         )
 
+    _ensure_api_flow_update_allowed(existing_flow, update_data)
+
     if settings_service.settings.remove_api_keys:
         update_data = remove_api_keys(update_data)
 
@@ -528,6 +545,8 @@ async def _patch_flow(
     silently move the flow into the actor's folder or write into the actor's
     fs namespace; the actor cannot change ownership-bound state at all.
     """
+    await lock_flow_for_update(session, db_flow)
+
     settings_service = get_settings_service()
 
     owner_user_id: UUID = db_flow.user_id
@@ -541,6 +560,8 @@ async def _patch_flow(
     if _endpoint_name_was_explicitly_cleared(flow):
         update_data["endpoint_name"] = None
 
+    _ensure_api_flow_update_allowed(db_flow, update_data)
+
     # A non-owner editing a shared flow must not be able to relocate the
     # flow into folders or storage they own. Reject ownership-bound mutations
     # explicitly so the failure surfaces in the response instead of silently

src/backend/base/langflow/services/database/models/flow/guards.py

@@ -0,0 +1,48 @@
+"""Application-level guards for locked flow mutations."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
+
+    from sqlmodel.ext.asyncio.session import AsyncSession
+
+    from langflow.services.database.models.flow.model import Flow
+
+LOCKED_FLOW_DETAIL = "Flow is locked. Unlock it before making changes."
+
+
+class LockedFlowError(RuntimeError):
+    """Raised when a mutation targets a locked flow."""
+
+
+async def lock_flow_for_update(session: AsyncSession, flow: Flow) -> None:
+    """Refresh *flow* while holding its database row lock until transaction end."""
+    await session.refresh(flow, with_for_update=True)
+
+
+def ensure_flow_unlocked(flow: Flow) -> None:
+    """Raise when *flow* is currently locked."""
+    if getattr(flow, "locked", False) is True:
+        raise LockedFlowError(LOCKED_FLOW_DETAIL)
+
+
+def ensure_flow_update_allowed(flow: Flow, update_data: Mapping[str, Any]) -> None:
+    """Allow updates to unlocked flows and unlock-only updates to locked flows.
+
+    API clients commonly send the full current flow when toggling the lock. We
+    therefore compare payload values with the persisted row and allow the
+    request when ``locked=False`` is the only effective change.
+    """
+    if getattr(flow, "locked", False) is not True:
+        return
+
+    changed_fields = {
+        field_name for field_name, new_value in update_data.items() if getattr(flow, field_name, None) != new_value
+    }
+    if changed_fields == {"locked"} and update_data.get("locked") is False:
+        return
+
+    raise LockedFlowError(LOCKED_FLOW_DETAIL)

src/backend/tests/unit/agentic/mcp/test_run_assistant_tool.py

@@ -159,6 +159,80 @@ async def test_should_persist_canvas_changes_on_an_existing_flow(self):
         assert result["flow_changed"] is True
         assert result["result"] == "Built a simple chat flow."
 
+    async def test_should_reject_locked_flow_before_running_assistant(self):
+        from langflow.agentic.utils.assistant_runner import run_assistant_and_persist
+        from langflow.services.database.models.flow.guards import LockedFlowError
+
+        user_id = uuid4()
+        flow = SimpleNamespace(
+            id=uuid4(),
+            name="Locked Flow",
+            data={"nodes": [], "edges": []},
+            user_id=user_id,
+            locked=True,
+        )
+        session = AsyncMock()
+        session.get = AsyncMock(return_value=flow)
+
+        with (
+            patch(f"{RUNNER_MODULE}._resolve_assistant_context", new_callable=AsyncMock) as resolve_context,
+            patch(f"{RUNNER_MODULE}.execute_flow_with_validation_streaming") as execute_stream,
+            pytest.raises(LockedFlowError, match="Flow is locked"),
+        ):
+            await run_assistant_and_persist(
+                session=session,
+                user_id=user_id,
+                instruction="Modify this flow",
+                flow_id=str(flow.id),
+            )
+
+        resolve_context.assert_not_awaited()
+        execute_stream.assert_not_called()
+
+    async def test_should_reject_flow_locked_during_assistant_execution(self):
+        from langflow.agentic.utils.assistant_runner import run_assistant_and_persist
+        from langflow.services.database.models.flow.guards import LockedFlowError
+
+        user_id = uuid4()
+        flow = SimpleNamespace(
+            id=uuid4(),
+            name="Unlocked Flow",
+            data={"nodes": [], "edges": []},
+            user_id=user_id,
+            locked=False,
+        )
+        session = AsyncMock()
+        session.get = AsyncMock(return_value=flow)
+
+        async def acquire_lock(*_args, **_kwargs):
+            flow.locked = True
+
+        session.refresh = AsyncMock(side_effect=acquire_lock)
+
+        with (
+            patch(
+                f"{RUNNER_MODULE}._resolve_assistant_context",
+                new_callable=AsyncMock,
+                return_value=_context_stub(),
+            ),
+            patch(
+                f"{RUNNER_MODULE}.execute_flow_with_validation_streaming",
+                side_effect=_stream_of(EVENTS_WITH_FLOW),
+            ),
+            patch(f"{RUNNER_MODULE}._save_flow_to_fs", new_callable=AsyncMock) as save_flow,
+            pytest.raises(LockedFlowError, match="Flow is locked"),
+        ):
+            await run_assistant_and_persist(
+                session=session,
+                user_id=user_id,
+                instruction="Modify this flow",
+                flow_id=str(flow.id),
+            )
+
+        session.refresh.assert_awaited_once_with(flow, with_for_update=True)
+        session.commit.assert_not_awaited()
+        save_flow.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_should_not_touch_the_flow_when_assistant_only_answers_text(self):
         from langflow.agentic.utils.assistant_runner import run_assistant_and_persist

src/backend/tests/unit/agentic/utils/test_flow_component.py

@@ -286,6 +286,39 @@ async def mock_scope():
         assert result["success"] is False
         assert "permission" in result["error"].lower()
 
+    async def test_should_reject_update_when_fresh_database_row_is_locked(self):
+        """The direct MCP DB path must honor a lock acquired after lookup."""
+        flow = _make_flow()
+        flow.locked = False
+        db_flow = MagicMock()
+        db_flow.user_id = UUID(USER_ID)
+        db_flow.locked = True
+
+        mock_session = AsyncMock()
+        mock_session.get = AsyncMock(return_value=db_flow)
+
+        @asynccontextmanager
+        async def mock_scope():
+            yield mock_session
+
+        with (
+            patch(f"{MODULE}.get_flow_by_id_or_endpoint_name", new_callable=AsyncMock, return_value=flow),
+            patch(f"{MODULE}.session_scope", mock_scope),
+            patch(f"{MODULE}.logger", _mock_logger()),
+        ):
+            result = await update_component_field_value(
+                "test-flow",
+                "comp-1",
+                "input_value",
+                "blocked value",
+                USER_ID,
+            )
+
+        assert result["success"] is False
+        assert result["error"] == "Flow is locked. Unlock it before making changes."
+        mock_session.refresh.assert_awaited_once_with(db_flow, with_for_update=True)
+        mock_session.commit.assert_not_awaited()
+
 
 class TestListComponentFields:
     """Tests for list_component_fields."""

src/backend/tests/unit/api/v1/test_flows.py

@@ -207,6 +207,70 @@ async def test_update_flow(client: AsyncClient, logged_in_headers):
     assert result["name"] == updated_name, "The name must be updated"
 
 
+async def test_locked_flow_rejects_api_updates_until_unlocked(client: AsyncClient, logged_in_headers):
+    original_data = {"nodes": [], "edges": []}
+    create_response = await client.post(
+        "api/v1/flows/",
+        json={"name": "locked-flow", "description": "original", "data": original_data, "locked": True},
+        headers=logged_in_headers,
+    )
+    assert create_response.status_code == status.HTTP_201_CREATED
+    created = create_response.json()
+    flow_id = created["id"]
+
+    patch_response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"description": "changed via PATCH"},
+        headers=logged_in_headers,
+    )
+    assert patch_response.status_code == status.HTTP_423_LOCKED
+    assert patch_response.json()["detail"] == "Flow is locked. Unlock it before making changes."
+
+    put_response = await client.put(
+        f"api/v1/flows/{flow_id}",
+        json={"name": "changed-via-put", "description": "original", "data": original_data},
+        headers=logged_in_headers,
+    )
+    assert put_response.status_code == status.HTTP_423_LOCKED
+
+    combined_unlock_response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"locked": False, "description": "changed while unlocking"},
+        headers=logged_in_headers,
+    )
+    assert combined_unlock_response.status_code == status.HTTP_423_LOCKED
+
+    unchanged_response = await client.get(f"api/v1/flows/{flow_id}", headers=logged_in_headers)
+    assert unchanged_response.status_code == status.HTTP_200_OK
+    assert unchanged_response.json()["name"] == "locked-flow"
+    assert unchanged_response.json()["description"] == "original"
+
+    # The UI sends the current flow fields along with the lock toggle. Equal
+    # values must not prevent an unlock-only request from succeeding.
+    unlock_response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={
+            "name": created["name"],
+            "description": created["description"],
+            "data": created["data"],
+            "folder_id": created["folder_id"],
+            "endpoint_name": created["endpoint_name"],
+            "locked": False,
+        },
+        headers=logged_in_headers,
+    )
+    assert unlock_response.status_code == status.HTTP_200_OK
+    assert unlock_response.json()["locked"] is False
+
+    update_response = await client.patch(
+        f"api/v1/flows/{flow_id}",
+        json={"description": "changed after unlock"},
+        headers=logged_in_headers,
+    )
+    assert update_response.status_code == status.HTTP_200_OK
+    assert update_response.json()["description"] == "changed after unlock"
+
+
 async def test_patch_flow_keeps_existing_endpoint_when_not_provided(client: AsyncClient, logged_in_headers):
     """Test that PATCH preserves endpoint_name when the field is omitted."""
     initial_flow = {