fix(api): stop the sync ceiling from capping background runs

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

@@ -462,6 +462,11 @@ async def _source(*, job_id=None, resume=None, **_kwargs):
                 # ``Job.result`` — protocol-neutral, so agui-protocol runs get a
                 # populated GET-status result too (not just langflow).
                 emit_output_capture=True,
+                # No client is waiting on a background run, so the sync HTTP ceiling is the wrong
+                # budget for it: nesting it inside the runner's asyncio.wait_for capped every
+                # background job at workflow_execution_timeout and made the documented
+                # background_job_timeout=None ("no timeout") unreachable. JobRunner owns it.
+                execution_timeout=None,
             ):
                 if terminal_error_type is not None and event_type == terminal_error_type:
                     errored = True

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

@@ -23,6 +23,7 @@
 import time
 from collections.abc import AsyncIterator
 from copy import deepcopy
+from typing import Final
 from uuid import UUID, uuid4
 
 from ag_ui.core import CustomEvent
@@ -71,6 +72,19 @@ def _resolve_execution_timeout() -> int:
 _EVENT_QUEUE_MAX_SIZE = 256
 
 
+class _CeilingFromSettings:
+    """Sentinel type for ``_stream_event_frames(execution_timeout=...)``.
+
+    Its own class rather than a bare ``object()`` so the parameter carries a real
+    static type and an ``isinstance`` check narrows the remaining value to
+    ``float | None`` for ``asyncio.wait_for``. Distinct from ``None`` so a caller
+    can ask for "unbounded" without it collapsing into "use the default".
+    """
+
+
+_CEILING_FROM_SETTINGS: Final = _CeilingFromSettings()
+
+
 async def generate_flow_events(*args, **kwargs) -> None:
     """Lazily call the v1 build stream to avoid import cycles during router setup."""
     from langflow.api.build import generate_flow_events as _generate_flow_events
@@ -187,6 +201,7 @@ async def _stream_event_frames(
     resume: dict | None = None,
     track_job_status: bool = True,
     emit_output_capture: bool = False,
+    execution_timeout: float | None | _CeilingFromSettings = _CEILING_FROM_SETTINGS,
 ) -> AsyncIterator[tuple[bytes, str]]:
     """Run a flow via the v1 build-vertex loop, dispatch its events through ``adapter``.
 
@@ -204,6 +219,14 @@ async def _stream_event_frames(
     the raw Langflow payload alongside the AG-UI translation for the
     playground's chat-view. A follow-up retires this once chat-view
     consumes the AG-UI ``TEXT_MESSAGE_*`` lifecycle directly.
+
+    ``execution_timeout`` bounds the run. It defaults to the settings ceiling,
+    which is the right budget for a caller with a waiting HTTP client (stream,
+    public). Background runs pass ``None``: nothing is waiting on them, and their
+    budget is ``background_job_timeout``, enforced by ``JobRunner`` one layer out.
+    Resolving the ceiling here for them too nested two budgets, and the inner one
+    always wins, which made the documented ``background_job_timeout=None``
+    ("no timeout") silently cap at the sync ceiling instead.
     """
     # EventManager uses put_nowait(), so a plain bounded asyncio.Queue would
     # silently drop frames via QueueFull. This adapter keeps memory bounded and
@@ -212,9 +235,11 @@ async def _stream_event_frames(
     event_manager = create_default_event_manager(queue)
     input_request = _single_input_value_request(parsed)
     flow_data = FlowDataRequest(**parsed.data) if parsed.data else None
-    # Single wall-clock ceiling for every mode that drives this loop (stream,
-    # background, public). Sync uses its own asyncio.wait_for upstream.
-    execution_timeout = _resolve_execution_timeout()
+    # Ceiling for the modes whose caller is waiting on a socket (stream, public).
+    # Sync uses its own asyncio.wait_for upstream; background passes None and is
+    # bounded by JobRunner instead. wait_for(timeout=None) simply awaits.
+    if isinstance(execution_timeout, _CeilingFromSettings):
+        execution_timeout = _resolve_execution_timeout()
 
     # Captured from drive()'s exception path so the consumer can yield a
     # guaranteed adapter.error_events(...) fallback after the queue loop ends.

src/backend/tests/unit/background_execution/test_execution_timeout_scope.py

@@ -0,0 +1,110 @@
+"""The sync HTTP ceiling must not bound a background run; JobRunner must.
+
+Before the fix, ``_stream_event_frames`` always applied ``workflow_execution_timeout``,
+so it nested inside the runner's ``asyncio.wait_for`` and won. That capped every
+background job at the sync ceiling and made the documented
+``background_job_timeout=None`` ("no timeout") unreachable.
+
+Real JobService, real migrated SQLite, real dispatcher, real runner. Only the ceiling
+and the work duration are shrunk so the tests stay fast.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from langflow.api.v2 import workflow_execution as wf_exec
+from langflow.services.background_execution.live_bus import InMemoryLiveBus
+from langflow.services.background_execution.runner import JobRunner
+from langflow.services.database.models.jobs.model import JobStatus
+from lfx.workflow.adapters import StreamAdapterContext, get_stream_adapter
+from lfx.workflow.converters import ParsedWorkflowRun
+
+
+def _source(adapter, *, mode: str, background: bool):
+    """Drive the real dispatcher.
+
+    ``background=True`` mirrors the production background call site, which passes
+    ``execution_timeout=None``.
+    """
+    extra = {"execution_timeout": None} if background else {}
+
+    async def _run(**_kwargs):
+        async for frame, event_type in wf_exec._stream_event_frames(
+            adapter=adapter,
+            flow_id=uuid4(),
+            flow_name="flow",
+            background_tasks=SimpleNamespace(add_task=lambda *_a, **_k: None),
+            parsed=ParsedWorkflowRun(flow_id=str(uuid4()), input_value="hi", mode=mode),
+            current_user=SimpleNamespace(id=uuid4()),
+            **extra,
+        ):
+            yield frame, event_type
+
+    return _run
+
+
+def _slow_build(work_s: float):
+    async def _build(**kwargs):
+        await asyncio.sleep(work_s)
+        queue = kwargs["event_manager"].queue
+        queue.put_nowait(("end", json.dumps({"event": "end", "data": {}}).encode(), time.time()))
+        await queue.put((None, None, time.time()))
+
+    return _build
+
+
+async def _run_job(job_service, *, ceiling, work, job_timeout, monkeypatch, background=True, mode="background"):
+    job_id = uuid4()
+    await job_service.create_job(job_id=job_id, flow_id=uuid4(), user_id=uuid4())
+    monkeypatch.setattr(wf_exec, "_resolve_execution_timeout", lambda: ceiling)
+    monkeypatch.setattr(wf_exec, "generate_flow_events", _slow_build(work))
+    adapter = get_stream_adapter("langflow", StreamAdapterContext(run_id=str(job_id), thread_id="t"))
+    runner = JobRunner(
+        job_service=job_service,
+        live_bus=InMemoryLiveBus(),
+        adapter=adapter,
+        frame_source=_source(adapter, mode=mode, background=background),
+        job_timeout=job_timeout,
+    )
+    await runner.run(job_id=job_id, source_kwargs={"job_id": job_id})
+    return await job_service.get_job_by_job_id(job_id)
+
+
+@pytest.mark.real_services
+@pytest.mark.no_blockbuster
+async def test_background_run_outlives_the_sync_ceiling(real_services_job_service, monkeypatch) -> None:
+    """background_job_timeout=None means no timeout, even past the sync ceiling."""
+    job = await _run_job(real_services_job_service, ceiling=0.05, work=0.5, job_timeout=None, monkeypatch=monkeypatch)
+    assert job.status == JobStatus.COMPLETED
+    assert job.error is None
+
+
+@pytest.mark.real_services
+@pytest.mark.no_blockbuster
+async def test_background_job_timeout_now_governs(real_services_job_service, monkeypatch) -> None:
+    """With background_job_timeout set, the runner bounds the run and marks it TIMED_OUT."""
+    job = await _run_job(real_services_job_service, ceiling=60.0, work=5.0, job_timeout=0.05, monkeypatch=monkeypatch)
+    assert job.status == JobStatus.TIMED_OUT
+
+
+@pytest.mark.real_services
+@pytest.mark.no_blockbuster
+async def test_stream_mode_still_enforces_the_ceiling(real_services_job_service, monkeypatch) -> None:
+    """Regression guard: a caller that does NOT opt out still gets the settings ceiling."""
+    job = await _run_job(
+        real_services_job_service,
+        ceiling=0.05,
+        work=5.0,
+        job_timeout=None,
+        monkeypatch=monkeypatch,
+        background=False,
+        mode="stream",
+    )
+    assert job.status == JobStatus.FAILED
+    assert "timed out" in str(job.error).lower()