From c23ff21d7ce80f7ec531ccbccc243236ecd89689 Mon Sep 17 00:00:00 2001
From: AKAZIK-py <200184714+AKAZIK-py@users.noreply.github.com>
Date: Fri, 31 Jul 2026 11:53:17 +0800
Subject: [PATCH] fix(moa): carry completed responses through the managed Relay
 path
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Under managed Relay execution the provider factory runs lazily inside
provider_stream() on the Relay session's event loop. The MoA facade's
auxiliary call_llm(stream=True) is invoked from that callback, so the
eager final_response check added for the non-managed path never fires:
the inner ManagedLlmStream is returned to the outer stream, which then
synchronously iterates it on the same loop thread and dies with
RuntimeError: Cannot run the event loop while another loop is running
(the completed response effectively trapped one level deeper).

stream_current() now detects a running event loop and returns the raw
factory result instead of nesting a ManagedLlmStream: the outer managed
stream already provides Relay tracking for the enclosing attempt, and
its own completed_response_predicate traps the completed response as
final_response — the same contract the main streaming worker consumes
(chat_completion_helpers reads stream.final_response after the chunk
loop). Nested managed streams remain supported for genuinely streaming
providers via the outer stream's own iteration.

Adds managed-execution regressions using the retained relay_turn
fixture: direct completed-response trapping, and the nested
facade-shaped stream_current call.
---
 agent/relay_llm.py            | 19 +++++++++
 tests/agent/test_relay_llm.py | 74 +++++++++++++++++++++++++++++++++++
 2 files changed, 93 insertions(+)

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"