fix(warm-registry): defer v2 host selection to settings (env-file safe); enforce exec timeout on warm sync runs

Author: Janardan S Kavia

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

@@ -1,6 +1,4 @@
 # Router for base api
-import os
-
 from fastapi import APIRouter, Depends
 from lfx.schema.workflow import WORKFLOW_EXECUTION_RESPONSES
 from lfx.services.settings.feature_flags import FEATURE_FLAGS
@@ -49,8 +47,7 @@
 from langflow.api.v2 import registration_router as registration_router_v2
 from langflow.api.v2 import workflow_background_router as workflow_background_router_v2
 from langflow.api.v2 import workflow_public_router as workflow_public_router_v2
-from langflow.api.v2.warm_workflow_host import WarmWorkflowHost
-from langflow.api.v2.workflow_host import LangflowWorkflowHost
+from langflow.api.v2.host_selection import DeferredWorkflowHost
 
 router_v1 = APIRouter(
     prefix="/v1",
@@ -155,15 +152,13 @@ def _include_agentic_router():
 # authenticated langflow v2 router has never carried a developer-api gate; the
 # default-off setting would otherwise 403 every authenticated request.
 # In PROD (``LANGFLOW_PROD``, execution-plane / ``--backend-only``) the warm host
-# serves deployed flows from the in-memory registry and skips per-flow RBAC. Read
-# the env directly rather than the settings service: this runs at import time,
-# before the settings service is guaranteed initialized. The value matches
-# ``settings.prod`` (env_prefix ``LANGFLOW_`` maps ``LANGFLOW_PROD`` -> ``prod``).
-_PROD = os.getenv("LANGFLOW_PROD", "").strip().lower() in ("1", "true", "yes", "on")
-# The one line that picks the runtime: warm host in PROD, DB-backed host otherwise.
-# Everything below is identical — only the host object swaps (the point of the seam).
-_workflow_host: WorkflowHost = WarmWorkflowHost() if _PROD else LangflowWorkflowHost()
-# Startup invariant: whichever host we picked satisfies the WorkflowHost protocol.
+# serves deployed flows from the in-memory registry and skips per-flow RBAC. The
+# warm-vs-DB choice is made per ``settings.prod`` but DEFERRED to first use: this
+# module is imported before ``load_dotenv(--env-file)`` runs, so reading the env
+# (or the settings service) here would miss ``--env-file`` values. The route
+# structure is identical for both hosts, so binding the deferred proxy changes
+# nothing structurally — only which concrete host each request resolves to.
+_workflow_host: WorkflowHost = DeferredWorkflowHost()
 assert isinstance(_workflow_host, WorkflowHost)  # noqa: S101
 router_v2.include_router(
     create_workflow_router(

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

@@ -0,0 +1,136 @@
+"""Deferred selection of the v2 workflow host (warm PROD registry vs DB-backed).
+
+Which host serves ``POST /api/v2/workflows`` depends on ``settings.prod``
+(``LANGFLOW_PROD``). The problem: ``langflow.__main__`` imports the router module
+(and therefore builds this router) *before* it runs ``load_dotenv(--env-file)``, so
+reading the env — or the settings service — at import time would miss any value
+supplied via ``--env-file``. That is exactly the ordering trap the extensions router
+documents for ``LANGFLOW_ENABLE_EXTENSION_RELOAD``.
+
+``DeferredWorkflowHost`` defers the choice to first use. By the time any workflow
+route (or capability flag) is exercised, ``setup_app``/lifespan has initialized the
+settings service from the fully-loaded environment, so ``settings.prod`` is correct.
+The route *structure* is identical for both hosts (the shared router only reads
+``supports_*`` at request time, and ``auto_register_job_routes=False`` neutralizes
+the one mount-time read), so binding this single proxy at import changes nothing
+structurally — only which concrete host each request lands on.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from lfx.workflow.host import WorkflowHostBase
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from fastapi import BackgroundTasks, Request, Response
+    from lfx.schema.workflow import (
+        ParsedWorkflowRun,
+        WorkflowExecutionResponse,
+        WorkflowJobResponse,
+        WorkflowStopResponse,
+    )
+    from lfx.workflow.host import ResolvedFlow, WorkflowAction
+
+
+class DeferredWorkflowHost(WorkflowHostBase):
+    """A ``WorkflowHost`` that picks the concrete host lazily from ``settings.prod``.
+
+    Every member delegates to the resolved host. Resolution is cached only once the
+    settings service is initialized, so an access during module import (before
+    ``--env-file`` is loaded) never force-initializes settings nor caches a stale
+    choice — it falls back transiently and re-resolves on the next call.
+    """
+
+    def __init__(self) -> None:
+        self._host: WorkflowHostBase | None = None
+
+    def _resolve(self) -> WorkflowHostBase:
+        if self._host is not None:
+            return self._host
+        from langflow.api.v2.warm_workflow_host import WarmWorkflowHost
+        from langflow.api.v2.workflow_host import LangflowWorkflowHost
+        from langflow.services.deps import get_settings_service, is_settings_service_initialized
+
+        # Do NOT force-initialize settings from a half-loaded environment at import
+        # time (the router is built before ``load_dotenv(--env-file)``). Until the
+        # settings service exists, fall back to the DB host WITHOUT caching so the
+        # real choice is still made on the first post-startup call.
+        if not is_settings_service_initialized():
+            return LangflowWorkflowHost()
+        host: WorkflowHostBase = WarmWorkflowHost() if get_settings_service().settings.prod else LangflowWorkflowHost()
+        self._host = host
+        return host
+
+    @property
+    def supports_background(self) -> bool:
+        return self._resolve().supports_background
+
+    @property
+    def supports_request_overrides(self) -> bool:
+        return self._resolve().supports_request_overrides
+
+    async def resolve_caller(self, request: Request) -> Any:
+        return await self._resolve().resolve_caller(request)
+
+    async def get_flow(self, flow_id: str, caller: Any) -> ResolvedFlow:
+        return await self._resolve().get_flow(flow_id, caller)
+
+    async def authorize(self, caller: Any, flow: ResolvedFlow, action: WorkflowAction) -> None:
+        return await self._resolve().authorize(caller, flow, action)
+
+    def session(self) -> AsyncIterator[Any | None]:
+        # Returns the concrete host's async context manager (used as ``async with``).
+        return self._resolve().session()
+
+    async def run_sync(
+        self,
+        parsed: ParsedWorkflowRun,
+        flow: ResolvedFlow,
+        caller: Any,
+        *,
+        http_request: Request,
+        background_tasks: BackgroundTasks,
+    ) -> WorkflowExecutionResponse:
+        return await self._resolve().run_sync(
+            parsed, flow, caller, http_request=http_request, background_tasks=background_tasks
+        )
+
+    def stream_response(
+        self,
+        parsed: ParsedWorkflowRun,
+        flow: ResolvedFlow,
+        caller: Any,
+        *,
+        stream_protocol: str,
+        http_request: Request,
+        background_tasks: BackgroundTasks,
+    ) -> Response:
+        return self._resolve().stream_response(
+            parsed,
+            flow,
+            caller,
+            stream_protocol=stream_protocol,
+            http_request=http_request,
+            background_tasks=background_tasks,
+        )
+
+    async def submit_background(
+        self,
+        parsed: ParsedWorkflowRun,
+        flow: ResolvedFlow,
+        caller: Any,
+        *,
+        stream_protocol: str,
+    ) -> WorkflowJobResponse:
+        return await self._resolve().submit_background(parsed, flow, caller, stream_protocol=stream_protocol)
+
+    async def get_job_status(
+        self, job_id: str, caller: Any, session: Any
+    ) -> WorkflowExecutionResponse | WorkflowJobResponse:
+        return await self._resolve().get_job_status(job_id, caller, session)
+
+    async def stop_job(self, job_id: str, caller: Any) -> WorkflowStopResponse:
+        return await self._resolve().stop_job(job_id, caller)

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

@@ -21,6 +21,7 @@
 
 from __future__ import annotations
 
+import asyncio
 from copy import deepcopy
 from typing import TYPE_CHECKING, Any
 
@@ -32,7 +33,8 @@
 from lfx.workflow.host import ResolvedFlow, WorkflowHostBase
 
 if TYPE_CHECKING:
-    from fastapi import Request
+    from fastapi import BackgroundTasks, Request
+    from lfx.schema.workflow import ParsedWorkflowRun, WorkflowExecutionResponse
 
 
 class WarmWorkflowHost(WorkflowHostBase):
@@ -113,3 +115,40 @@ async def get_flow(self, flow_id: str, caller: Any) -> ResolvedFlow:  # noqa: AR
         # Return a run-ready Graph (not a FlowRead) as ResolvedFlow, so the base
         # run_sync/stream can execute it directly — no from_payload rebuild on this path.
         return ResolvedFlow(flow_id=flow_id, graph=graph_copy, session_id_default=flow_id)
+
+    async def run_sync(
+        self,
+        parsed: ParsedWorkflowRun,
+        flow: ResolvedFlow,
+        caller: Any,
+        *,
+        http_request: Request,
+        background_tasks: BackgroundTasks,
+    ) -> WorkflowExecutionResponse:
+        """Run the graph with the langflow wall-clock ceiling enforced (408 on timeout).
+
+        The lean base ``run_sync`` has no timeout, so a runaway flow on the execution
+        plane would run unbounded and hold a worker forever. Wrap it in the configured
+        ``workflow_execution_timeout`` and surface the same 408 contract the DB host
+        uses. (Only per-flow RBAC + job/vertex-build writes stay dropped; the durable
+        job-tracking and header-global/session mapping remain a separate adapter.)
+        """
+        from langflow.services.deps import get_settings_service
+
+        timeout_seconds = get_settings_service().settings.workflow_execution_timeout
+        try:
+            return await asyncio.wait_for(
+                super().run_sync(parsed, flow, caller, http_request=http_request, background_tasks=background_tasks),
+                timeout=timeout_seconds,
+            )
+        except (TimeoutError, asyncio.TimeoutError):
+            raise HTTPException(
+                status_code=status.HTTP_408_REQUEST_TIMEOUT,
+                detail={
+                    "error": "Execution timeout",
+                    "code": "EXECUTION_TIMEOUT",
+                    "message": f"Workflow execution exceeded {timeout_seconds} seconds",
+                    "flow_id": flow.flow_id,
+                    "timeout_seconds": timeout_seconds,
+                },
+            ) from None

src/backend/tests/unit/services/warm_registry/test_warm_registry_reconcile.py

@@ -505,3 +505,73 @@ async def _boom_once():
     with contextlib.suppress(asyncio.CancelledError):
         await task
     assert calls >= 1
+
+
+# ── #2: deferred host selection (fixes import-time-vs-env-file ordering) ──────
+def test_deferred_host_resolves_db_when_not_prod(monkeypatch):
+    """settings.prod False -> DB-backed LangflowWorkflowHost."""
+    from types import SimpleNamespace
+
+    from langflow.api.v2.host_selection import DeferredWorkflowHost
+    from langflow.api.v2.workflow_host import LangflowWorkflowHost
+    from langflow.services import deps
+
+    monkeypatch.setattr(deps, "is_settings_service_initialized", lambda: True)
+    monkeypatch.setattr(deps, "get_settings_service", lambda: SimpleNamespace(settings=SimpleNamespace(prod=False)))
+    host = DeferredWorkflowHost()
+    assert isinstance(host._resolve(), LangflowWorkflowHost)
+
+
+def test_deferred_host_resolves_warm_when_prod(monkeypatch):
+    """settings.prod True -> WarmWorkflowHost, and the choice is cached."""
+    from types import SimpleNamespace
+
+    from langflow.api.v2.host_selection import DeferredWorkflowHost
+    from langflow.api.v2.warm_workflow_host import WarmWorkflowHost
+    from langflow.services import deps
+
+    monkeypatch.setattr(deps, "is_settings_service_initialized", lambda: True)
+    monkeypatch.setattr(deps, "get_settings_service", lambda: SimpleNamespace(settings=SimpleNamespace(prod=True)))
+    host = DeferredWorkflowHost()
+    resolved = host._resolve()
+    assert isinstance(resolved, WarmWorkflowHost)
+    assert host._resolve() is resolved  # cached
+
+
+def test_deferred_host_does_not_cache_before_settings_init(monkeypatch):
+    """An access before settings are initialized (import time) must not cache a choice."""
+    from langflow.api.v2.host_selection import DeferredWorkflowHost
+    from langflow.api.v2.workflow_host import LangflowWorkflowHost
+    from langflow.services import deps
+
+    monkeypatch.setattr(deps, "is_settings_service_initialized", lambda: False)
+    host = DeferredWorkflowHost()
+    resolved = host._resolve()
+    assert isinstance(resolved, LangflowWorkflowHost)  # transient DB fallback
+    assert host._host is None  # NOT cached — real choice deferred to first post-startup call
+
+
+# ── #5: warm host enforces the workflow_execution_timeout (408) on sync runs ──
+async def test_warm_host_run_sync_enforces_timeout(monkeypatch):
+    """A sync run exceeding workflow_execution_timeout -> HTTP 408 (the lean base has none)."""
+    from types import SimpleNamespace
+
+    import lfx.workflow.router as lfx_router
+    from fastapi import HTTPException
+    from langflow.api.v2.warm_workflow_host import WarmWorkflowHost
+    from langflow.services import deps
+
+    async def _slow(*_args, **_kwargs):
+        await asyncio.sleep(1)
+
+    monkeypatch.setattr(lfx_router, "run_workflow_sync", _slow)
+    monkeypatch.setattr(
+        deps, "get_settings_service", lambda: SimpleNamespace(settings=SimpleNamespace(workflow_execution_timeout=0.01))
+    )
+
+    host = WarmWorkflowHost()
+    flow = SimpleNamespace(graph=object(), flow_id="fid")
+    with pytest.raises(HTTPException) as exc:
+        await host.run_sync(SimpleNamespace(), flow, None, http_request=None, background_tasks=None)
+    assert exc.value.status_code == 408
+    assert exc.value.detail["code"] == "EXECUTION_TIMEOUT"