diff --git a/src/backend/base/langflow/agentic/services/flow_executor.py b/src/backend/base/langflow/agentic/services/flow_executor.py index 3499239ef5..30d2ce5047 100644 --- a/src/backend/base/langflow/agentic/services/flow_executor.py +++ b/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 diff --git a/src/backend/tests/unit/agentic/services/test_flow_executor_uses_coordinator.py b/src/backend/tests/unit/agentic/services/test_flow_executor_uses_coordinator.py new file mode 100644 index 0000000000..56fcb13baa --- /dev/null +++ b/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