diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 2d3a2883e173f..96a98d54d24b5 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -247,6 +247,14 @@ async def execute_current_async( ) +def _has_running_event_loop() -> bool: + try: + asyncio.get_running_loop() + except RuntimeError: + return False + return True + + def stream_current( request: dict[str, Any], stream_factory: Callable[[dict[str, Any]], Any], @@ -272,6 +280,17 @@ def stream_current( turn = relay_runtime.active_turn() if turn is None: return stream_factory(request) + if _has_running_event_loop(): + # Managed provider callbacks execute on the Relay session's event + # loop. A nested ManagedLlmStream built here would be synchronously + # iterated on that same loop thread, which asyncio forbids + # ("Cannot run the event loop while another loop is running"). + # Return the raw factory result instead: the outer managed stream + # already provides Relay tracking for the enclosing attempt, and its + # own completed_response_predicate traps a completed response (e.g. + # the MoA facade's auxiliary ``call_llm(stream=True)`` returning a + # full response when an adapter ignores ``stream=True``). + return stream_factory(request) managed = stream( request, stream_factory, diff --git a/tests/agent/test_relay_llm.py b/tests/agent/test_relay_llm.py index e506fad1e88ba..e24897412a7c7 100644 --- a/tests/agent/test_relay_llm.py +++ b/tests/agent/test_relay_llm.py @@ -795,3 +795,77 @@ def test_stream_current_streams_iterators_with_predicate(tmp_path, monkeypatch): relay_runtime.SESSION_COORDINATOR.release_conversation(lease) relay_runtime._reset_for_tests() + + +def _completed_response(content: str = "done") -> SimpleNamespace: + return SimpleNamespace( + model="test-model", + choices=[ + SimpleNamespace( + message=SimpleNamespace( + role="assistant", + content=content, + tool_calls=None, + ), + finish_reason="stop", + ) + ], + usage=None, + ) + + +def _choices_predicate(value) -> bool: + return hasattr(value, "choices") + + +def test_stream_managed_traps_direct_completed_response(relay_turn): + """Managed path: a factory returning a completed response (adapter + ignoring stream=True) is trapped as final_response instead of iterated.""" + relay, turn = relay_turn + del relay, turn + + stream = relay_llm.stream( + {"model": "test-model", "messages": []}, + lambda request: _completed_response(), + session_id="session-1", + name="test-provider", + model_name="test-model", + finalizer=lambda: {}, + completed_response_predicate=_choices_predicate, + ) + assert list(stream) == [] + assert stream.final_response is not None + assert stream.final_response.choices[0].message.content == "done" + + +def test_stream_current_inside_managed_callback_returns_raw(relay_turn): + """Managed path: an auxiliary stream_current() call made from inside a + managed provider callback (the MoA facade's call_llm(stream=True) shape) + must return the raw factory result; the outer stream traps a completed + response as its final_response instead of crashing on a nested event + loop or surfacing an empty stream.""" + relay, turn = relay_turn + del relay, turn + + def outer_factory(request): + return relay_llm.stream_current( + {"model": "test-model", "messages": []}, + lambda inner_request: _completed_response(), + name="moa-aggregator", + model_name="test-model", + finalizer=lambda: {}, + completed_response_predicate=_choices_predicate, + ) + + stream = relay_llm.stream( + {"model": "test-model", "messages": []}, + outer_factory, + session_id="session-1", + name="moa", + model_name="test-model", + finalizer=lambda: {}, + completed_response_predicate=_choices_predicate, + ) + assert list(stream) == [] + assert stream.final_response is not None + assert stream.final_response.choices[0].message.content == "done"