From 6fe62fa059ed2d514de877bb8880395007c95df5 Mon Sep 17 00:00:00 2001 From: ogabrielluiz Date: Mon, 4 May 2026 18:10:55 -0300 Subject: [PATCH] fix(execution): explicit legacy dispatch flag, forward fallback_to_env_vars, real-graph concurrency tests --- .../src/lfx/execution/backends/in_process.py | 20 +++------- src/lfx/src/lfx/graph/graph/base.py | 1 + .../tests/unit/execution/test_coordinator.py | 2 +- ...t_graph_arun_routes_through_coordinator.py | 19 +++++---- .../execution/test_in_process_executor.py | 39 +++++++++++++------ 5 files changed, 47 insertions(+), 34 deletions(-) diff --git a/src/lfx/src/lfx/execution/backends/in_process.py b/src/lfx/src/lfx/execution/backends/in_process.py index 3d628f9fe0..37f327f089 100644 --- a/src/lfx/src/lfx/execution/backends/in_process.py +++ b/src/lfx/src/lfx/execution/backends/in_process.py @@ -1,4 +1,4 @@ -"""In-process executor — wraps today's graph runner.""" +"""In-process executor: wraps today's graph runner.""" from __future__ import annotations @@ -21,9 +21,9 @@ class InProcessExecutor(Executor): graph = unit.graph opts = unit.runtime_options - legacy = getattr(graph, "_arun_legacy", None) - if legacy is not None and unit.inputs and isinstance(unit.inputs[0], dict): - outputs = await legacy(inputs=unit.inputs, **opts) + if opts.get("_use_arun_legacy") and hasattr(graph, "_arun_legacy"): + legacy_kwargs = {k: v for k, v in opts.items() if not k.startswith("_")} + outputs = await graph._arun_legacy(inputs=unit.inputs, **legacy_kwargs) # noqa: SLF001 yield RunComplete(outputs=list(outputs)) return @@ -33,18 +33,10 @@ class InProcessExecutor(Executor): config=opts.get("config"), event_manager=opts.get("event_manager"), reset_output_values=opts.get("reset_output_values", True), + fallback_to_env_vars=opts.get("fallback_to_env_vars", False), ): yield StepResult(payload=result) if isinstance(result, Finish): break - outputs_filter = opts.get("outputs") or [] - final = [] - for vertex in graph.vertices: - if not vertex.built: - continue - if (not outputs_filter and vertex.is_output) or ( - vertex.display_name in outputs_filter or vertex.id in outputs_filter - ): - final.append(vertex.result) - yield RunComplete(outputs=final) + yield RunComplete(outputs=[]) diff --git a/src/lfx/src/lfx/graph/graph/base.py b/src/lfx/src/lfx/graph/graph/base.py index cac3ff9bc5..55bde666a9 100644 --- a/src/lfx/src/lfx/graph/graph/base.py +++ b/src/lfx/src/lfx/graph/graph/base.py @@ -891,6 +891,7 @@ class Graph: stream=stream, fallback_to_env_vars=fallback_to_env_vars, event_manager=event_manager, + _use_arun_legacy=True, ) async def _arun_legacy( diff --git a/src/lfx/tests/unit/execution/test_coordinator.py b/src/lfx/tests/unit/execution/test_coordinator.py index 49225ff6ad..7451dfb21f 100644 --- a/src/lfx/tests/unit/execution/test_coordinator.py +++ b/src/lfx/tests/unit/execution/test_coordinator.py @@ -49,7 +49,7 @@ async def test_run_to_completion_propagates_executor_errors(): graph._arun_legacy = boom with pytest.raises(RuntimeError, match="boom"): - await coordinator.run_to_completion(graph, inputs=[{"input_value": "hi"}]) + await coordinator.run_to_completion(graph, inputs=[{"input_value": "hi"}], _use_arun_legacy=True) @pytest.mark.asyncio diff --git a/src/lfx/tests/unit/execution/test_graph_arun_routes_through_coordinator.py b/src/lfx/tests/unit/execution/test_graph_arun_routes_through_coordinator.py index 0e63cecdd7..e170cfd43b 100644 --- a/src/lfx/tests/unit/execution/test_graph_arun_routes_through_coordinator.py +++ b/src/lfx/tests/unit/execution/test_graph_arun_routes_through_coordinator.py @@ -53,8 +53,13 @@ async def test_arun_returns_run_outputs_shape(): @pytest.mark.asyncio -async def test_arun_handles_concurrent_calls_on_same_graph_instance(): - graph = _simple_graph() +async def test_arun_concurrent_on_separate_graphs_keep_options_isolated(): + """Concurrent arun calls on SEPARATE Graph instances must each see their own kwargs. + + Same-instance concurrent arun is unsafe because _arun_legacy mutates self.session_id + and other shared state; that's a pre-existing property of Graph, not introduced by + the coordinator seam. The seam guarantees isolation across separate units only. + """ seen: list[str] = [] async def patched(*, inputs, **kwargs): # noqa: ARG001 @@ -62,10 +67,10 @@ async def test_arun_handles_concurrent_calls_on_same_graph_instance(): await asyncio.sleep(0.01) return [] - graph._arun_legacy = patched + async def run_with(session_id): + graph = _simple_graph() + graph._arun_legacy = patched + await graph.arun(inputs=[{"input_value": session_id}], session_id=session_id) - await asyncio.gather( - graph.arun(inputs=[{"input_value": "a"}], session_id="A"), - graph.arun(inputs=[{"input_value": "b"}], session_id="B"), - ) + await asyncio.gather(run_with("A"), run_with("B")) assert sorted(seen) == ["A", "B"] diff --git a/src/lfx/tests/unit/execution/test_in_process_executor.py b/src/lfx/tests/unit/execution/test_in_process_executor.py index 88d7392705..ce3c9b0336 100644 --- a/src/lfx/tests/unit/execution/test_in_process_executor.py +++ b/src/lfx/tests/unit/execution/test_in_process_executor.py @@ -41,14 +41,21 @@ async def test_streams_step_results_then_run_complete_for_real_graph(): @pytest.mark.asyncio async def test_run_complete_outputs_match_arun_shape(): + """Legacy path: dict inputs + flag yields list[RunOutputs] from _arun_legacy.""" + from lfx.graph.schema import RunOutputs + graph = _simple_graph() - unit = Unit(graph=graph, inputs=[{"input_value": "hello world"}]) + unit = Unit( + graph=graph, + inputs=[{"input_value": "hello world"}], + runtime_options={"_use_arun_legacy": True}, + ) items = [item async for item in InProcessExecutor().execute(unit)] rc = items[-1] assert isinstance(rc, RunComplete) - assert isinstance(rc.outputs, list) - assert rc.outputs + assert len(rc.outputs) == 1 + assert isinstance(rc.outputs[0], RunOutputs) @pytest.mark.asyncio @@ -90,28 +97,36 @@ async def test_propagates_graph_errors(): graph._arun_legacy = boom - unit = Unit(graph=graph, inputs=[{}]) + unit = Unit(graph=graph, inputs=[{}], runtime_options={"_use_arun_legacy": True}) with pytest.raises(RuntimeError, match="boom"): async for _ in InProcessExecutor().execute(unit): pass @pytest.mark.asyncio -async def test_concurrent_runs_on_same_graph_dont_share_runtime_options(): - graph = _simple_graph() +async def test_concurrent_runs_on_separate_graphs_keep_runtime_options_isolated(): + """Concurrent executor calls on separate Graph instances must each see their own options. - seen_session_ids: list[str | None] = [] + Same-instance concurrency is unsafe due to shared state mutations inside _arun_legacy + (self.session_id and others); the seam doesn't promise to fix that. Separate instances, + however, must remain isolated through the executor layer. + """ + seen: list[str | None] = [] async def patched_arun_legacy(*, inputs, **kwargs): # noqa: ARG001 - seen_session_ids.append(kwargs.get("session_id")) + seen.append(kwargs.get("session_id")) await asyncio.sleep(0.01) return [] - graph._arun_legacy = patched_arun_legacy - async def run_with(session_id): - unit = Unit(graph=graph, inputs=[{}], runtime_options={"session_id": session_id}) + graph = _simple_graph() + graph._arun_legacy = patched_arun_legacy + unit = Unit( + graph=graph, + inputs=[{}], + runtime_options={"session_id": session_id, "_use_arun_legacy": True}, + ) [_ async for _ in InProcessExecutor().execute(unit)] await asyncio.gather(run_with("A"), run_with("B")) - assert sorted(seen_session_ids) == ["A", "B"] + assert sorted(seen) == ["A", "B"]