feat(execution): route flow_executor through coordinator (seam #2)

src/backend/base/langflow/agentic/services/flow_executor.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any
 from fastapi import HTTPException
 from lfx.cli.script_loader import extract_structured_result
 from lfx.events.event_manager import EventManager, create_default_event_manager
+from lfx.execution import StepResult, get_default_coordinator
 from lfx.log.logger import logger
 from lfx.schema.schema import InputValueRequest
 from lfx.utils.flow_validation import CustomComponentValidationError
@@ -59,7 +60,13 @@ async def _run_graph_with_events(
         graph.prepare()
         inputs = InputValueRequest(input_value=input_value) if input_value else None
 
-        results = [result async for result in graph.async_start(inputs=inputs, event_manager=event_manager)]
+        results = [
+            item.payload
+            async for item in get_default_coordinator().run(
+                graph, inputs=[], initial_inputs=inputs, event_manager=event_manager
+            )
+            if isinstance(item, StepResult)
+        ]
         execution_result.result = extract_structured_result(results)
     except Exception as e:  # noqa: BLE001
         execution_result.error = e
@@ -131,7 +138,11 @@ async def execute_flow_file(
         graph.prepare()
         inputs = InputValueRequest(input_value=input_value) if input_value else None
 
-        results = [result async for result in graph.async_start(inputs=inputs)]
+        results = [
+            item.payload
+            async for item in get_default_coordinator().run(graph, inputs=[], initial_inputs=inputs)
+            if isinstance(item, StepResult)
+        ]
         flow_result = extract_structured_result(results)
     except HTTPException:
         raise

src/backend/tests/unit/agentic/services/test_flow_executor_uses_coordinator.py

@@ -0,0 +1,81 @@
+"""Verify flow_executor.py routes through the executor coordinator."""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any
+
+import pytest
+from lfx.execution import (
+    Coordinator,
+    ExecutorRegistry,
+    set_default_coordinator,
+)
+from lfx.execution.executor import Executor
+from lfx.execution.types import RunComplete, StepResult
+
+
+class _Recording(Executor):
+    kind = "in-process"
+
+    def __init__(self) -> None:
+        self.units_seen: list[Any] = []
+
+    async def execute(self, unit):
+        self.units_seen.append(unit)
+        # Yield a fake vertex result with no .vertex attr so extract_structured_result
+        # falls through to the default "no response" branch (still a dict).
+        yield StepResult(payload=object())
+        yield RunComplete(outputs=[])
+
+
+@pytest.fixture(autouse=True)
+def _reset_singleton():
+    from lfx import execution
+
+    execution._default_registry = None
+    execution._default_coordinator = None
+    yield
+    execution._default_registry = None
+    execution._default_coordinator = None
+
+
+@pytest.mark.asyncio
+async def test_run_graph_uses_default_coordinator():
+    """Coordinator dispatch on _run_graph_with_events."""
+    from langflow.agentic.services.flow_executor import _run_graph_with_events
+    from langflow.agentic.services.flow_types import FlowExecutionResult
+
+    recording = _Recording()
+    registry = ExecutorRegistry()
+    registry.register(recording)
+    set_default_coordinator(Coordinator(registry=registry))
+
+    class _FakeGraph:
+        flow_id: str | None = None
+        flow_name: str | None = None
+        user_id: str | None = None
+        session_id: str | None = None
+        context: dict = {}
+
+        def prepare(self):
+            pass
+
+    queue: asyncio.Queue = asyncio.Queue()
+    execution_result = FlowExecutionResult()
+
+    from lfx.events.event_manager import create_default_event_manager
+
+    em = create_default_event_manager(asyncio.Queue())
+    await _run_graph_with_events(
+        graph=_FakeGraph(),
+        input_value="hi",
+        global_variables=None,
+        user_id=None,
+        session_id=None,
+        event_manager=em,
+        event_queue=queue,
+        execution_result=execution_result,
+    )
+
+    assert len(recording.units_seen) == 1