diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 610deff98f..270960fd26 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -286,6 +286,12 @@ def stream_current( own ``hasattr(stream, "choices")`` check handled it (#11732, #55933) — without the unwrap the response stays trapped as ``final_response`` on the inner ManagedLlmStream and the outer consumer sees an empty stream. + + Determining that return shape requires starting the lazy managed pipeline, + and Relay may read ahead internally while satisfying that first pull. A + genuine first returned chunk remains buffered, while provider work, + latency, and pre-first-yield errors may surface before this function + returns. """ turn = relay_runtime.active_turn() if turn is None: diff --git a/tests/agent/test_relay_llm.py b/tests/agent/test_relay_llm.py index 67dad0fc47..e28f28ea67 100644 --- a/tests/agent/test_relay_llm.py +++ b/tests/agent/test_relay_llm.py @@ -822,6 +822,36 @@ def test_stream_current_primes_lazy_completed_response(relay_turn, monkeypatch): assert result is completed +def test_stream_current_unwraps_completed_response_with_real_interceptor(relay_turn): + """A real stream interceptor makes Relay lazy; completion still unwraps.""" + relay, _turn = relay_turn + completed = _completed_response() + + async def identity_stream(request, next_call): + return await next_call(request) + + relay.intercepts.register_llm_stream_execution( + "hermes-test-prime-completed", + 1, + identity_stream, + ) + try: + result = relay_llm.stream_current( + {"model": "test-model", "messages": [], "stream": True}, + lambda _request: completed, + name="test-provider", + model_name="test-model", + finalizer=lambda: completed, + completed_response_predicate=_choices_predicate, + ) + + assert result is completed + finally: + relay.intercepts.deregister_llm_stream_execution( + "hermes-test-prime-completed" + ) + + def test_stream_current_preserves_real_relay_interceptor_chunks(relay_turn): """Priming a real managed pipeline must retain its transformed first chunk.""" relay, _turn = relay_turn @@ -860,6 +890,23 @@ def test_stream_current_preserves_real_relay_interceptor_chunks(relay_turn): ) +def test_stream_current_surfaces_managed_factory_error_before_return(relay_turn): + """Shape detection preserves the unmanaged factory-error boundary.""" + + def fail_factory(_request): + raise RuntimeError("provider failed before streaming") + + with pytest.raises(RuntimeError, match="provider failed before streaming"): + relay_llm.stream_current( + {"model": "test-model", "messages": [], "stream": True}, + fail_factory, + name="test-provider", + model_name="test-model", + finalizer=dict, + completed_response_predicate=_choices_predicate, + ) + + def _completed_response(content: str = "done") -> SimpleNamespace: return SimpleNamespace( model="test-model",