fix(initial_setup): preserve variable bindings when loading flows from disk (#14053)

fix(initial_setup): preserve variable bindings on flow load

Author: erichare

docs/docs/Develop/environment-variables.mdx

@@ -446,6 +446,7 @@ See [Telemetry](/contributing-telemetry).
 | `LANGFLOW_ALLOW_COMPONENTS_PATHS_OVERRIDE` | Boolean | `True` | When `false` alongside `LANGFLOW_ALLOW_CUSTOM_COMPONENTS=false`, components contributed by `LANGFLOW_COMPONENTS_PATH` and `LANGFLOW_COMPONENTS_INDEX_PATH` no longer bypass the block. Has no effect when `LANGFLOW_ALLOW_CUSTOM_COMPONENTS=true`. For more information, see [Block custom components](../Deployment/deployment-block-custom-components.mdx). |
 | `LANGFLOW_LOAD_FLOWS_PATH` | String | Not set | Path to a directory containing flow JSON files to be loaded on startup. Typically used when creating a Docker image with prepackaged flows. Requires `LANGFLOW_AUTO_LOGIN=True`. |
 | `LANGFLOW_LOAD_FLOWS_OVERWRITE_ON_NAME_MATCH` | Boolean | `False` | When a flow file in `LANGFLOW_LOAD_FLOWS_PATH` shares a name with an existing DB row but has a different `id`, controls whether to overwrite the existing row. `False` (default) skips with a warning so UI edits are preserved on restart when file UUIDs regenerate. Set to `True` to opt into prepackaged-flows-are-source-of-truth semantics, typically for CI/CD pipelines. |
+| `LANGFLOW_LOAD_FLOWS_PRESERVE_VARIABLE_BINDINGS` | Boolean | `True` | Preserve global-variable bindings configured in the visual editor when an existing flow is reloaded from `LANGFLOW_LOAD_FLOWS_PATH`. Bindings explicitly configured in the flow file still take precedence. Set to `False` to restore blind replacement of the flow data on restart. |
 | `LANGFLOW_CREATE_STARTER_PROJECTS` | Boolean | `True` | Whether to create templates during initialization. If `false`, Langflow doesn't create templates, and `LANGFLOW_UPDATE_STARTER_PROJECTS` is treated as `false`. |
 | `LANGFLOW_UPDATE_STARTER_PROJECTS` | Boolean | `True` | Whether to update templates with the latest component versions when initializing after an upgrade. |
 | `LANGFLOW_LAZY_LOAD_COMPONENTS` | Boolean | `False` | If `true`, Langflow only partially loads components at startup and fully loads them on demand. This significantly reduces startup time but can cause a slight delay when a component is first used. |
@@ -485,4 +486,4 @@ You can hide individual elements without enabling the umbrella flag.
 | `LANGFLOW_HIDE_NEW_PROJECT_BUTTON` | Boolean | `False` | If `true`, hides the new project/folder button in the sidebar. Automatically enabled when `LANGFLOW_EMBEDDED_MODE=true`. |
 | `LANGFLOW_HIDE_NEW_FLOW_BUTTON` | Boolean | `False` | If `true`, hides the new flow button in the header. Automatically enabled when `LANGFLOW_EMBEDDED_MODE=true`. |
 | `LANGFLOW_HIDE_STARTER_PROJECTS` | Boolean | `False` | If `true`, hides the starter projects tab in the templates modal. Does not affect database seeding of starter projects. Automatically enabled when `LANGFLOW_EMBEDDED_MODE=true`. |
-| `LANGFLOW_HIDE_GETTING_STARTED_PROGRESS` | Boolean | `False` | If `true`, hides the getting-started onboarding progress UI. Not automatically enabled when `LANGFLOW_EMBEDDED_MODE=true`. |
+| `LANGFLOW_HIDE_GETTING_STARTED_PROGRESS` | Boolean | `False` | If `true`, hides the getting-started onboarding progress UI. Not automatically enabled when `LANGFLOW_EMBEDDED_MODE=true`. |

src/backend/base/langflow/initial_setup/setup.py

@@ -1120,6 +1120,82 @@ async def load_bundles_from_urls() -> tuple[list[TemporaryDirectory], list[str]]
 )
 
 
+def _get_component_data(node):
+    if not isinstance(node, dict):
+        return None
+    node_data = node.get("data")
+    if not isinstance(node_data, dict):
+        return None
+    component_data = node_data.get("node")
+    return component_data if isinstance(component_data, dict) else None
+
+
+def _get_node_template(node):
+    component_data = _get_component_data(node)
+    if component_data is None:
+        return None
+    template = component_data.get("template")
+    return template if isinstance(template, dict) else None
+
+
+def _get_nested_flow(node):
+    component_data = _get_component_data(node)
+    if component_data is None:
+        return None
+    nested_flow = component_data.get("flow")
+    return nested_flow if isinstance(nested_flow, dict) else None
+
+
+def _is_variable_binding(field):
+    if not isinstance(field, dict) or field.get("load_from_db") is not True:
+        return False
+    variable_name = field.get("value")
+    return isinstance(variable_name, str) and bool(variable_name)
+
+
+def _merge_variable_bindings(existing_data, incoming_data):
+    """Preserve DB-backed field bindings while taking flow structure from the incoming file."""
+    merged_data = deepcopy(incoming_data)
+    if not isinstance(existing_data, dict) or not isinstance(merged_data, dict):
+        return merged_data
+
+    existing_nodes = existing_data.get("nodes")
+    incoming_nodes = merged_data.get("nodes")
+    if not isinstance(existing_nodes, list) or not isinstance(incoming_nodes, list):
+        return merged_data
+
+    existing_nodes_by_id = {
+        node["id"]: node for node in existing_nodes if isinstance(node, dict) and isinstance(node.get("id"), str)
+    }
+    for incoming_node in incoming_nodes:
+        if not isinstance(incoming_node, dict):
+            continue
+        node_id = incoming_node.get("id")
+        if not isinstance(node_id, str) or node_id not in existing_nodes_by_id:
+            continue
+
+        existing_node = existing_nodes_by_id[node_id]
+        existing_template = _get_node_template(existing_node)
+        incoming_template = _get_node_template(incoming_node)
+        if existing_template is not None and incoming_template is not None:
+            for field_name, incoming_field in incoming_template.items():
+                if not isinstance(incoming_field, dict):
+                    continue
+                existing_field = existing_template.get(field_name)
+                if _is_variable_binding(existing_field) and not _is_variable_binding(incoming_field):
+                    incoming_field["value"] = deepcopy(existing_field["value"])
+                    incoming_field["load_from_db"] = True
+
+        existing_nested_flow = _get_nested_flow(existing_node)
+        incoming_nested_flow = _get_nested_flow(incoming_node)
+        if existing_nested_flow is not None and incoming_nested_flow is not None and "data" in incoming_nested_flow:
+            incoming_nested_flow["data"] = _merge_variable_bindings(
+                existing_nested_flow.get("data"), incoming_nested_flow["data"]
+            )
+
+    return merged_data
+
+
 async def upsert_flow_from_file(file_content: AnyStr, filename: str, session: AsyncSession, user_id: UUID) -> None:
     flow = orjson.loads(file_content)
     flow_endpoint_name = flow.get("endpoint_name")
@@ -1143,6 +1219,7 @@ async def upsert_flow_from_file(file_content: AnyStr, filename: str, session: As
         name=flow_name,
     )
     if existing:
+        settings = get_settings_service().settings
         await logger.adebug(f"Found existing flow: {existing.name}")
         # Normalize the DB id to UUID for comparison without mutating the attached
         # row: SQLAlchemy can return ids as strings on SQLite, but assigning back
@@ -1157,7 +1234,7 @@ async def upsert_flow_from_file(file_content: AnyStr, filename: str, session: As
         else:
             db_id = db_id_raw
         matched_by_id = flow_id is not None and db_id == flow_id
-        if not matched_by_id and not get_settings_service().settings.load_flows_overwrite_on_name_match:
+        if not matched_by_id and not settings.load_flows_overwrite_on_name_match:
             await logger.awarning(
                 f"Skipping flow update: db_id={db_id} name={existing.name!r} matched by "
                 f"name/endpoint_name but file id differs (file id={flow_id}). "
@@ -1174,7 +1251,10 @@ async def upsert_flow_from_file(file_content: AnyStr, filename: str, session: As
         # lazy load outside greenlet context and raises ``MissingGreenlet``.
         for key in _FLOW_UPDATABLE_COLUMNS:
             if key in flow:
-                setattr(existing, key, flow[key])
+                incoming_value = flow[key]
+                if key == "data" and settings.load_flows_preserve_variable_bindings:
+                    incoming_value = _merge_variable_bindings(existing.data, incoming_value)
+                setattr(existing, key, incoming_value)
         existing.updated_at = datetime.now(tz=timezone.utc).astimezone()
         existing.user_id = user_id
 

src/backend/tests/unit/initial_setup/test_upsert_flow_from_file.py

@@ -36,6 +36,46 @@ def _overwrite_on_name_match(*, enabled: bool):
         settings.load_flows_overwrite_on_name_match = original
 
 
+@contextmanager
+def _preserve_variable_bindings(*, enabled: bool):
+    """Temporarily override the load_flows_preserve_variable_bindings setting."""
+    settings = get_settings_service().settings
+    original = settings.load_flows_preserve_variable_bindings
+    settings.load_flows_preserve_variable_bindings = enabled
+    try:
+        yield
+    finally:
+        settings.load_flows_preserve_variable_bindings = original
+
+
+def _node_with_field(node_id: str, *, value: str, load_from_db: bool) -> dict:
+    return {
+        "id": node_id,
+        "data": {
+            "node": {
+                "template": {
+                    "api_key": {
+                        "value": value,
+                        "load_from_db": load_from_db,
+                    }
+                }
+            }
+        },
+    }
+
+
+def _group_node(node_id: str, *nodes: dict) -> dict:
+    return {
+        "id": node_id,
+        "data": {
+            "node": {
+                "template": {},
+                "flow": {"data": {"nodes": list(nodes), "edges": []}},
+            }
+        },
+    }
+
+
 async def _create_flow(
     *,
     name: str,
@@ -310,6 +350,162 @@ async def test_upsert_flow_from_file_id_match_still_overwrites_id_field() -> Non
         assert rows[0].description == "updated"
 
 
+@pytest.mark.usefixtures("client")
+@pytest.mark.parametrize("incoming_load_from_db", [False, True])
+async def test_upsert_flow_from_file_preserves_existing_variable_binding(incoming_load_from_db) -> None:
+    """A UI-configured global-variable binding survives a same-id startup re-import."""
+    user_id = uuid4()
+    existing_node = _node_with_field("n1", value="OPENAI_API_KEY", load_from_db=True)
+    existing_node["data"]["node"]["template"]["api_key"]["display_name"] = "Old label"
+    original = await _create_flow(
+        name="BoundFlow",
+        user_id=user_id,
+        data={"nodes": [existing_node], "edges": []},
+    )
+    incoming_data = {
+        "nodes": [_node_with_field("n1", value="", load_from_db=incoming_load_from_db)],
+        "edges": [{"id": "updated-edge"}],
+    }
+    incoming_data["nodes"][0]["data"]["node"]["template"]["api_key"]["display_name"] = "New label"
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=True):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        field = updated.data["nodes"][0]["data"]["node"]["template"]["api_key"]
+        assert field == {"value": "OPENAI_API_KEY", "load_from_db": True, "display_name": "New label"}
+        assert updated.data["edges"] == incoming_data["edges"]
+
+
+@pytest.mark.usefixtures("client")
+async def test_upsert_flow_from_file_explicit_file_variable_binding_wins() -> None:
+    """An explicit binding in the file remains the source of truth."""
+    user_id = uuid4()
+    original = await _create_flow(
+        name="ReboundFlow",
+        user_id=user_id,
+        data={"nodes": [_node_with_field("n1", value="OLD_API_KEY", load_from_db=True)], "edges": []},
+    )
+    incoming_data = {
+        "nodes": [_node_with_field("n1", value="NEW_API_KEY", load_from_db=True)],
+        "edges": [],
+    }
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=True):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        field = updated.data["nodes"][0]["data"]["node"]["template"]["api_key"]
+        assert field == {"value": "NEW_API_KEY", "load_from_db": True}
+
+
+@pytest.mark.usefixtures("client")
+async def test_upsert_flow_from_file_preserves_nested_variable_binding() -> None:
+    """Bindings inside grouped flows survive the same recursive startup merge."""
+    user_id = uuid4()
+    existing_group = _group_node("group", _node_with_field("nested", value="NESTED_API_KEY", load_from_db=True))
+    original = await _create_flow(
+        name="GroupedFlow",
+        user_id=user_id,
+        data={"nodes": [existing_group], "edges": []},
+    )
+    incoming_group = _group_node("group", _node_with_field("nested", value="", load_from_db=False))
+    incoming_data = {"nodes": [incoming_group], "edges": []}
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=True):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        nested_node = updated.data["nodes"][0]["data"]["node"]["flow"]["data"]["nodes"][0]
+        field = nested_node["data"]["node"]["template"]["api_key"]
+        assert field == {"value": "NESTED_API_KEY", "load_from_db": True}
+
+
+@pytest.mark.usefixtures("client")
+async def test_upsert_flow_from_file_keeps_new_file_nodes_unchanged() -> None:
+    """Nodes without a DB counterpart pass through the merge unchanged."""
+    user_id = uuid4()
+    original = await _create_flow(
+        name="ExpandedFlow",
+        user_id=user_id,
+        data={"nodes": [_node_with_field("existing", value="API_KEY", load_from_db=True)], "edges": []},
+    )
+    new_node = _node_with_field("new", value="literal-from-file", load_from_db=False)
+    incoming_data = {"nodes": [new_node], "edges": []}
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=True):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        assert updated.data == incoming_data
+
+
+@pytest.mark.usefixtures("client")
+async def test_upsert_flow_from_file_can_disable_variable_binding_preservation() -> None:
+    """The opt-out restores the previous blind-overwrite behavior."""
+    user_id = uuid4()
+    original = await _create_flow(
+        name="GitOpsFlow",
+        user_id=user_id,
+        data={"nodes": [_node_with_field("n1", value="API_KEY", load_from_db=True)], "edges": []},
+    )
+    incoming_data = {
+        "nodes": [_node_with_field("n1", value="literal-from-file", load_from_db=False)],
+        "edges": [],
+    }
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=False):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        assert updated.data == incoming_data
+
+
+@pytest.mark.usefixtures("client")
+async def test_upsert_flow_from_file_does_not_preserve_empty_default_binding() -> None:
+    """An empty load-from-DB default must not override a literal value from the file."""
+    user_id = uuid4()
+    original = await _create_flow(
+        name="DefaultSecretFlow",
+        user_id=user_id,
+        data={"nodes": [_node_with_field("n1", value="", load_from_db=True)], "edges": []},
+    )
+    incoming_data = {
+        "nodes": [_node_with_field("n1", value="literal-from-file", load_from_db=False)],
+        "edges": [],
+    }
+    file_content = orjson.dumps({"id": str(original.id), "name": original.name, "data": incoming_data})
+
+    with _preserve_variable_bindings(enabled=True):
+        async with session_scope() as session:
+            await upsert_flow_from_file(file_content, original.name, session, user_id)
+            await session.commit()
+
+    async with session_scope() as session:
+        updated = (await session.exec(select(Flow).where(Flow.id == original.id))).one()
+        assert updated.data == incoming_data
+
+
 @pytest.mark.usefixtures("client")
 async def test_upsert_flow_from_file_skips_name_match_when_overwrite_disabled() -> None:
     """When load_flows_overwrite_on_name_match=False, name-matched rows are NOT overwritten.

src/lfx/src/lfx/services/settings/groups/components.py

@@ -35,6 +35,12 @@ class ComponentsSettings(BaseModel):
     successfully either way.) Set ``True`` to opt into "prepackaged flows are the source of
     truth on restart" semantics, typically for CI/CD pipelines.
     """
+    load_flows_preserve_variable_bindings: bool = True
+    """Preserve global-variable bindings configured in the UI when reloading an existing flow.
+
+    The flow file remains authoritative for the rest of the flow structure and for bindings it
+    explicitly defines. Set this to ``False`` to restore blind ``data`` replacement on restart.
+    """
     bundle_urls: list[str] = []
 
     lazy_load_components: bool = False

src/lfx/tests/unit/services/settings/test_settings_composition.py

@@ -160,6 +160,7 @@
     "mcp_servers_locked",
     # ComponentsSettings
     "load_flows_overwrite_on_name_match",
+    "load_flows_preserve_variable_bindings",
     "enable_extension_reload",
     # SecuritySettings
     "rate_limit_enabled",
@@ -257,6 +258,7 @@ def test_critical_defaults_unchanged():
     assert settings.mcp_server_allowed_packages is None
     assert settings.mcp_server_enabled is True
     assert settings.mcp_composer_enabled is True
+    assert settings.load_flows_preserve_variable_bindings is True
     assert settings.do_not_track is False
     assert settings.dev is False
     assert settings.agentic_experience is False
@@ -389,6 +391,12 @@ def test_yaml_round_trip():
         ("LANGFLOW_BACKEND_ONLY", "true", "backend_only", True),
         ("LANGFLOW_AUTO_SAVING", "false", "auto_saving", False),
         ("LANGFLOW_FALLBACK_TO_ENV_VAR", "false", "fallback_to_env_var", False),
+        (
+            "LANGFLOW_LOAD_FLOWS_PRESERVE_VARIABLE_BINDINGS",
+            "false",
+            "load_flows_preserve_variable_bindings",
+            False,
+        ),
         ("LANGFLOW_VARIABLE_STORE", "kubernetes", "variable_store", "kubernetes"),
     ],
 )