diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 39ae9a759c3..c24cc972a2e 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -5711,6 +5711,8 @@ def call_llm( timeout: float = None, extra_body: dict = None, api_mode: str = None, + stream: bool = False, + stream_options: dict = None, ) -> Any: """Centralized synchronous LLM call. @@ -5731,9 +5733,16 @@ def call_llm( tools: Tool definitions (for function calling). timeout: Request timeout in seconds (None = read from auxiliary.{task}.timeout config). extra_body: Additional request body fields. + stream: When True, return the raw SDK streaming iterator instead of a + validated complete response. The caller is responsible for consuming + chunks (and for any fallback). Used by the MoA aggregator so its + output can stream to the user. + stream_options: Passed through to the request when stream is True + (e.g. {"include_usage": True}). Returns: - Response object with .choices[0].message.content + Response object with .choices[0].message.content, OR — when stream=True — + the raw streaming iterator from client.chat.completions.create(). Raises: RuntimeError: If no provider is configured. @@ -5835,6 +5844,20 @@ def call_llm( if _is_anthropic_compat_endpoint(resolved_provider, _client_base): kwargs["messages"] = _convert_openai_images_to_anthropic(kwargs["messages"]) + # Streaming path: return the raw SDK Stream iterator directly. This is used by + # the MoA aggregator so its tokens stream to the user. It deliberately skips + # _validate_llm_response and the temperature/max_tokens/payment fallback chain + # below — those all assume a complete response object, whereas a stream is + # consumed chunk-by-chunk by the caller. The caller (the agent's streaming + # consumer) owns chunk reassembly, stale-stream detection, and falling back to + # a non-streaming call on error. stream_options is best-effort: providers that + # reject it surface an error the caller's fallback already handles. + if stream: + kwargs["stream"] = True + if stream_options: + kwargs["stream_options"] = stream_options + return client.chat.completions.create(**kwargs) + # Handle unsupported temperature, max_tokens vs max_completion_tokens retry, # then payment fallback. try: diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index 4ca09c6895d..7a5919807af 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -1168,11 +1168,22 @@ def run_conversation( # stream. Mirror the ACP exclusion used for Responses # API upgrade (lines ~1083-1085). elif ( - agent.provider in {"copilot-acp", "moa"} + agent.provider in {"copilot-acp"} or str(agent.base_url or "").lower().startswith("acp://copilot") or str(agent.base_url or "").lower().startswith("acp+tcp://") ): _use_streaming = False + # MoA streams only when a display/TTS consumer is present to + # receive the deltas. MoAChatCompletions.create() honors + # stream=True (runs the references, then returns the aggregator's + # raw token stream) and is reached here because, for provider + # "moa", _create_request_openai_client returns the MoA facade + # itself. Without consumers (quiet mode, subagents, health-check + # probes) we keep the complete-response path: the facade returns a + # whole response when stream is not requested, preserving the + # prior behavior for those callers. + elif agent.provider == "moa" and not agent._has_stream_consumers(): + _use_streaming = False elif not agent._has_stream_consumers(): # No display/TTS consumer. Still prefer streaming for # health checking, but skip for Mock clients in tests diff --git a/agent/moa_loop.py b/agent/moa_loop.py index 583a0d56ccb..fcc76c2cf0f 100644 --- a/agent/moa_loop.py +++ b/agent/moa_loop.py @@ -577,6 +577,24 @@ class MoAChatCompletions: # max_tokens is passed through from the caller (normally None → omitted # → the model's real maximum). The preset's old hardcoded 4096 default # is gone — it truncated long syntheses. + # When the agent's streaming consumer calls us with stream=True, run the + # references first (above) and then return the aggregator's RAW token + # stream so the acting model's output reaches the user live. The consumer + # reassembles chunks + tool_calls, runs stale-stream detection, and falls + # back to a non-streaming retry on error. The non-streaming path + # (stream=False) is unchanged — no stream/stream_options/timeout are + # forwarded, so its behavior is byte-for-byte identical to before. + stream = bool(api_kwargs.get("stream")) + stream_kwargs: dict[str, Any] = {} + if stream: + stream_kwargs["stream"] = True + stream_kwargs["stream_options"] = ( + api_kwargs.get("stream_options") or {"include_usage": True} + ) + # Forward the consumer's per-request (stream read) timeout so it + # actually governs the aggregator stream, not just call_llm's default. + if api_kwargs.get("timeout") is not None: + stream_kwargs["timeout"] = api_kwargs["timeout"] return call_llm( task="moa_aggregator", messages=agg_messages, @@ -584,6 +602,7 @@ class MoAChatCompletions: max_tokens=agg_kwargs.get("max_tokens"), tools=agg_kwargs.get("tools"), extra_body=agg_kwargs.get("extra_body"), + **stream_kwargs, **_slot_runtime(aggregator), ) diff --git a/tests/run_agent/test_moa_streaming.py b/tests/run_agent/test_moa_streaming.py new file mode 100644 index 00000000000..a61780aca08 --- /dev/null +++ b/tests/run_agent/test_moa_streaming.py @@ -0,0 +1,221 @@ +"""Tests for MoA aggregator streaming. + +MoAChatCompletions.create() honors stream=True by running the references first +and then returning the aggregator's raw streaming iterator (from call_llm), so +the acting model's output can stream to the user. stream=False is the original +complete-response path and must stay byte-identical. +""" +from types import SimpleNamespace + +import pytest + + +def _response(content="done", *, tool_calls=None): + message = SimpleNamespace(content=content, tool_calls=tool_calls or []) + choice = SimpleNamespace(message=message, finish_reason="stop") + return SimpleNamespace(choices=[choice], usage=None, model="fake-model") + + +def _write_cfg(home): + home.mkdir() + (home / "config.yaml").write_text( + """ +moa: + default_preset: review + presets: + review: + reference_models: + - provider: openai-codex + model: gpt-5.5 + aggregator: + provider: openrouter + model: anthropic/claude-opus-4.8 +""".strip(), + encoding="utf-8", + ) + + +def _facade(monkeypatch, tmp_path, on_call=None): + home = tmp_path / ".hermes" + _write_cfg(home) + monkeypatch.setenv("HERMES_HOME", str(home)) + calls = [] + + def fake_call_llm(**kwargs): + calls.append(kwargs) + if on_call is not None: + r = on_call(kwargs) + if r is not None: + return r + if kwargs["task"] == "moa_reference": + return _response("reference advice") + return _response("aggregator acted") + + monkeypatch.setattr("agent.moa_loop.call_llm", fake_call_llm) + from agent.moa_loop import MoAChatCompletions + + return MoAChatCompletions("review"), calls + + +# -------------------------------------------------------------------------- +# Facade-level: create() stream branch +# -------------------------------------------------------------------------- + +def test_create_streams_aggregator_when_requested(monkeypatch, tmp_path): + """stream=True: references still run, aggregator is called with stream=True + and stream_options, and create() returns the aggregator call's result + (the raw stream) verbatim.""" + sentinel = object() + + def on_call(kwargs): + if kwargs["task"] == "moa_aggregator": + return sentinel + return None + + facade, calls = _facade(monkeypatch, tmp_path, on_call=on_call) + out = facade.create( + messages=[{"role": "user", "content": "q"}], + tools=[{"type": "function"}], + stream=True, + ) + + # create() returns the aggregator's streaming result untouched. + assert out is sentinel + # References still ran (MoA not bypassed). + assert any(c["task"] == "moa_reference" for c in calls) + agg = next(c for c in calls if c["task"] == "moa_aggregator") + assert agg["stream"] is True + assert agg["stream_options"] == {"include_usage": True} + # Tools still flow to the (streaming) aggregator. + assert agg["tools"] is not None + + +def test_create_non_stream_path_unchanged(monkeypatch, tmp_path): + """Default (no stream): the aggregator call carries NO stream/stream_options + keys, so the non-streaming path is byte-identical to before.""" + facade, calls = _facade(monkeypatch, tmp_path) + facade.create(messages=[{"role": "user", "content": "q"}], tools=[]) + + agg = next(c for c in calls if c["task"] == "moa_aggregator") + assert "stream" not in agg + assert "stream_options" not in agg + assert "timeout" not in agg + + +def test_create_forwards_stream_read_timeout(monkeypatch, tmp_path): + """The consumer's per-request (stream read) timeout is forwarded to the + aggregator so it actually governs the stream.""" + timeout_sentinel = object() + facade, calls = _facade(monkeypatch, tmp_path) + facade.create( + messages=[{"role": "user", "content": "q"}], + tools=[], + stream=True, + timeout=timeout_sentinel, + ) + agg = next(c for c in calls if c["task"] == "moa_aggregator") + assert agg["timeout"] is timeout_sentinel + + +def test_create_respects_caller_stream_options(monkeypatch, tmp_path): + """A caller-provided stream_options is forwarded as-is (not overwritten).""" + facade, calls = _facade(monkeypatch, tmp_path) + facade.create( + messages=[{"role": "user", "content": "q"}], + tools=[], + stream=True, + stream_options={"include_usage": False, "extra": 1}, + ) + agg = next(c for c in calls if c["task"] == "moa_aggregator") + assert agg["stream_options"] == {"include_usage": False, "extra": 1} + + +def test_create_does_not_forward_timeout_when_not_streaming(monkeypatch, tmp_path): + """A stray timeout on a non-streaming call is NOT forwarded — the non-stream + path must remain unchanged regardless of incidental kwargs.""" + facade, calls = _facade(monkeypatch, tmp_path) + facade.create(messages=[{"role": "user", "content": "q"}], tools=[], timeout=object()) + agg = next(c for c in calls if c["task"] == "moa_aggregator") + assert "timeout" not in agg + assert "stream" not in agg + + +# -------------------------------------------------------------------------- +# call_llm-level: stream branch returns the raw SDK stream +# -------------------------------------------------------------------------- + +def test_call_llm_stream_returns_raw_stream_and_skips_validation(monkeypatch): + """call_llm(stream=True) returns the client's raw stream object directly, + attaches stream/stream_options to the request, and does NOT run response + validation (which assumes a complete response).""" + from agent import auxiliary_client as ac + + captured = {} + + class _Completions: + def create(self, **kwargs): + captured.update(kwargs) + return "RAW_STREAM" + + fake_client = SimpleNamespace( + chat=SimpleNamespace(completions=_Completions()), + base_url="http://localhost:8001/v1", + ) + + monkeypatch.setattr( + ac, "_resolve_task_provider_model", + lambda *a, **k: ("custom", "m", "http://localhost:8001/v1", "key", "chat_completions"), + ) + monkeypatch.setattr(ac, "_get_cached_client", lambda *a, **k: (fake_client, "m")) + + def _no_validate(*a, **k): + raise AssertionError("streaming must not go through _validate_llm_response") + + monkeypatch.setattr(ac, "_validate_llm_response", _no_validate) + + out = ac.call_llm( + provider="custom", + model="m", + messages=[{"role": "user", "content": "hi"}], + stream=True, + stream_options={"include_usage": True}, + ) + + assert out == "RAW_STREAM" + assert captured.get("stream") is True + assert captured.get("stream_options") == {"include_usage": True} + + +def test_call_llm_non_stream_still_validates(monkeypatch): + """Sanity: stream=False keeps the validated path (regression guard for the + early-return not leaking into normal calls).""" + from agent import auxiliary_client as ac + + class _Completions: + def create(self, **kwargs): + return _response("ok") + + fake_client = SimpleNamespace( + chat=SimpleNamespace(completions=_Completions()), + base_url="http://localhost:8001/v1", + ) + monkeypatch.setattr( + ac, "_resolve_task_provider_model", + lambda *a, **k: ("custom", "m", "http://localhost:8001/v1", "key", "chat_completions"), + ) + monkeypatch.setattr(ac, "_get_cached_client", lambda *a, **k: (fake_client, "m")) + + validated = {"called": False} + + def _validate(resp, task): + validated["called"] = True + return resp + + monkeypatch.setattr(ac, "_validate_llm_response", _validate) + + ac.call_llm( + provider="custom", + model="m", + messages=[{"role": "user", "content": "hi"}], + ) + assert validated["called"] is True