feat(preload): Introduce per-step completion flags for improved state management

- Added completion flags in the preload state to track the status of various initialization steps, including profile picture copying, starter project creation, agentic global variable initialization, MCP server configuration, and flow loading.
- Updated the lifespan management in `main.py` to utilize these flags, allowing the system to skip redundant setup tasks if they have already been completed during the preload phase.
- This enhancement improves resource management and ensures that the application behaves correctly in a multi-worker environment.

src/backend/base/langflow/main.py

@@ -153,7 +153,7 @@ def get_lifespan(*, fix_migration=False, version=None):
     async def lifespan(_app: FastAPI):
         from lfx.interface.components import get_and_cache_all_types_dict
 
-        from langflow.preload import get_preloaded_temp_dirs, is_master, is_preloaded
+        from langflow.preload import _STATE, get_preloaded_temp_dirs, is_master, is_preloaded
 
         preloaded = is_preloaded()
         running_in_master = is_master()
@@ -208,7 +208,7 @@ def get_lifespan(*, fix_migration=False, version=None):
             setup_llm_caching()
             await logger.adebug(f"LLM caching setup in {asyncio.get_event_loop().time() - current_time:.2f}s")
 
-            if not preloaded:
+            if not preloaded or not _STATE.profile_pictures_copied:
                 current_time = asyncio.get_event_loop().time()
                 await logger.adebug("Copying profile pictures")
                 await copy_profile_pictures()
@@ -225,10 +225,8 @@ def get_lifespan(*, fix_migration=False, version=None):
                 await logger.adebug("Initializing super user")
                 await initialize_auto_login_default_superuser()
                 await logger.adebug(f"Super user initialized in {asyncio.get_event_loop().time() - current_time:.2f}s")
-            else:
-                await logger.adebug(
-                    "Skipping profile-picture copy and super user setup: master already ran them during preload"
-                )
+            elif _STATE.profile_pictures_copied:
+                await logger.adebug("Skipping profile-picture copy: master already completed it during preload")
 
             if get_settings_service().settings.prometheus_enabled:
                 try:
@@ -255,7 +253,7 @@ def get_lifespan(*, fix_migration=False, version=None):
 
             telemetry_service = get_telemetry_service()
 
-            if preloaded:
+            if preloaded and _STATE.starter_projects_created:
                 # Inherit bundle paths and types dict from master via COW.
                 # Only the master owns the bundle temp dirs; workers must NOT
                 # add them to their own cleanup list, or shutdown would delete
@@ -299,7 +297,7 @@ def get_lifespan(*, fix_migration=False, version=None):
                     )
 
                 # Initialize agentic global variables early (before MCP server and flows)
-                if get_settings_service().settings.agentic_experience:
+                if get_settings_service().settings.agentic_experience and not _STATE.agentic_globals_initialized:
                     from langflow.api.utils.mcp.agentic_mcp import initialize_agentic_global_variables
 
                     current_time = asyncio.get_event_loop().time()
@@ -328,8 +326,8 @@ def get_lifespan(*, fix_migration=False, version=None):
             )
 
             # Auto-configure Agentic MCP server if enabled (after variables are initialized).
-            # Skipped when preloaded: the master already seeded this into the DB.
-            if get_settings_service().settings.agentic_experience and not preloaded:
+            # Skipped when the master already completed this during preload.
+            if get_settings_service().settings.agentic_experience and not _STATE.agentic_mcp_configured:
                 from langflow.api.utils.mcp.agentic_mcp import auto_configure_agentic_mcp_server
 
                 current_time = asyncio.get_event_loop().time()
@@ -344,9 +342,11 @@ def get_lifespan(*, fix_migration=False, version=None):
                     await logger.awarning(f"Failed to configure agentic MCP server: {e}")
 
             current_time = asyncio.get_event_loop().time()
-            if not preloaded:
+            if not _STATE.flows_loaded:
                 await logger.adebug("Loading flows")
                 await load_flows_from_directory()
+            else:
+                await logger.adebug("Skipping flows load: master already completed it during preload")
             # ``sync_flows_from_fs`` and the queue service MUST be started
             # per-worker: they create asyncio tasks bound to this event loop.
             sync_flows_from_fs_task = asyncio.create_task(sync_flows_from_fs())

src/backend/base/langflow/preload.py

@@ -46,6 +46,12 @@ class _PreloadState:
     master_pid: int | None = None
     temp_dirs: list[TemporaryDirectory] = field(default_factory=list)
     bundles_components_paths: list[str] = field(default_factory=list)
+    # Per-step completion flags to prevent silent data loss
+    profile_pictures_copied: bool = False
+    starter_projects_created: bool = False
+    agentic_globals_initialized: bool = False
+    agentic_mcp_configured: bool = False
+    flows_loaded: bool = False
 
 
 _STATE = _PreloadState()
@@ -106,6 +112,7 @@ async def _run_master_preload() -> None:
     await logger.adebug("[preload] copying profile pictures")
     try:
         await copy_profile_pictures()
+        _STATE.profile_pictures_copied = True
     except Exception as e:  # noqa: BLE001
         await logger.awarning(f"[preload] copy_profile_pictures failed: {e}")
 
@@ -123,6 +130,7 @@ async def _run_master_preload() -> None:
         await logger.adebug("[preload] creating/updating starter projects")
         try:
             await create_or_update_starter_projects(all_types_dict)
+            _STATE.starter_projects_created = True
         except Exception as e:  # noqa: BLE001
             await logger.awarning(f"[preload] starter projects init failed: {e}")
 
@@ -136,15 +144,19 @@ async def _run_master_preload() -> None:
             await logger.adebug("[preload] initializing agentic global variables")
             async with session_scope() as session:
                 await initialize_agentic_global_variables(session)
+            _STATE.agentic_globals_initialized = True
+
             await logger.adebug("[preload] auto-configuring agentic MCP server")
             async with session_scope() as session:
                 await auto_configure_agentic_mcp_server(session)
+            _STATE.agentic_mcp_configured = True
         except Exception as e:  # noqa: BLE001
             await logger.awarning(f"[preload] agentic MCP init failed: {e}")
 
     await logger.adebug("[preload] loading flows from directory")
     try:
         await load_flows_from_directory()
+        _STATE.flows_loaded = True
     except Exception as e:  # noqa: BLE001
         await logger.awarning(f"[preload] load_flows_from_directory failed: {e}")