fix(execution): explicit legacy dispatch flag, forward fallback_to_env_vars, real-graph concurrency tests

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=[])

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(

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

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"]

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"]