From 5121a2a20e47701025374612d41cdec29116ec96 Mon Sep 17 00:00:00 2001 From: kudi88 Date: Sat, 25 Jul 2026 12:49:17 -0700 Subject: [PATCH] feat(aux): force streaming for providers that reject non-stream requests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Some OpenAI-compatible endpoints — notably Tencent Copilot (copilot.tencent.com) — only accept streaming chat requests; any non-streaming call returns HTTP 400 (code 11101, 'Non-stream chat request is currently not supported'). The main conversation loop already streams, so interactive chat works, but every auxiliary task (title generation, compression, web extraction) used the non-streaming path and failed on each call. _provider_requires_stream() detects stream-only endpoints (copilot.tencent.com built in, plus user-configurable auxiliary.stream_only_base_urls substring markers in config.yaml). Matching sync auxiliary calls route through _create_with_progress (force_stream=True) and async calls through the new _acreate_with_stream, aggregating the chunk stream — including tool-call deltas and reasoning deltas — into a complete response via the shared _ChatStreamAccumulator. Salvaged from PR #60686 by @kudi88 onto the progress-aware streaming machinery from #71508, addressing both sweeper-review gaps: the async path now consumes the stream with 'async for' (awaiting create() and iterating synchronously raised on AsyncOpenAI streams), and tool-call deltas are reassembled instead of dropped (MCP passes tools= through call_llm). Under force_stream there is no silent non-streaming retry — a stream-only provider rejects those by definition, so the original error surfaces to the normal recovery chains. --- agent/auxiliary_client.py | 332 +++++++++++++++------ contributors/emails/kudi3699@gmail.com | 1 + hermes_cli/config.py | 7 + tests/agent/test_aux_progress_streaming.py | 146 ++++++++- website/docs/user-guide/configuration.md | 12 + 5 files changed, 404 insertions(+), 94 deletions(-) create mode 100644 contributors/emails/kudi3699@gmail.com diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 7e4b9ae43b32..74ccd5781297 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -7234,20 +7234,67 @@ def _is_streaming_rejected_error(exc: Exception) -> bool: ) -def _create_with_progress(client: Any, kwargs: Dict[str, Any], task: Optional[str] = None) -> Any: - """chat.completions.create() that streams when a progress hook is active. +def _provider_requires_stream(provider: str, base_url: Optional[str]) -> bool: + """Detect providers that only accept streaming (non-stream = HTTP 400). + + Some OpenAI-compatible endpoints reject non-streaming chat requests + outright — e.g. Tencent Copilot returns + ``{"code": 11101, "msg": "Non-stream chat request is currently not + supported"}``. The main conversation loop already streams, so interactive + chat works; auxiliary tasks (title generation, compression, web extract) + used the non-streaming path and failed on every call. When this returns + True the auxiliary client sends ``stream=True`` and aggregates the chunks + itself (see :func:`_aggregate_chat_stream`). Credit @kudi88 (PR #60686). + + Beyond the known-host list, users can mark ANY custom endpoint as + stream-only via ``auxiliary.stream_only_base_urls`` in config.yaml + (list of substrings matched against the endpoint URL). + """ + _url = str(base_url or "").lower() + if not _url: + return False + # Tencent Copilot — "Non-stream chat request is currently not supported" + if base_url_host_matches(_url, "copilot.tencent.com"): + return True + try: + from hermes_cli.config import load_config + aux_cfg = (load_config() or {}).get("auxiliary", {}) + markers = aux_cfg.get("stream_only_base_urls") or [] + if isinstance(markers, (list, tuple)): + for marker in markers: + if isinstance(marker, str) and marker.strip() and marker.strip().lower() in _url: + return True + except Exception: + # Config read is best-effort; never break an aux call over it. + pass + return False + + +def _create_with_progress( + client: Any, + kwargs: Dict[str, Any], + task: Optional[str] = None, + *, + force_stream: bool = False, +) -> Any: + """chat.completions.create() that streams when a progress hook is active + or the provider only accepts streamed requests. Behavior is byte-for-byte identical to a plain ``create(**kwargs)`` when - no hook is installed (every existing caller/task) or when the client's + neither trigger applies (every existing caller/task) or when the client's wire adapter streams internally. With a hook + a chunk-capable client, the request is sent with ``stream=True`` and aggregated, ticking the hook per chunk — so the configured ``timeout`` acts per stream read (idle) rather than as a total budget, and outer liveness watchdogs see tokens - moving. Providers that reject the streamed request fall back to the - plain non-streaming call. + moving. ``force_stream=True`` (stream-only providers such as Tencent + Copilot — credit @kudi88, PR #60686) takes the same streamed path even + without a hook. Providers that reject the streamed request fall back to + the plain non-streaming call — except under ``force_stream``, where a + stream-only provider rejects the plain call by definition, so the + original error is surfaced to the normal recovery chains instead. """ _notify_aux_progress() # request dispatched counts as progress - if not _aux_progress_active() or _client_streams_internally(client): + if (not _aux_progress_active() and not force_stream) or _client_streams_internally(client): return client.chat.completions.create(**kwargs) total_ceiling = _aux_stream_total_ceiling(kwargs.get("timeout")) @@ -7262,7 +7309,8 @@ def _create_with_progress(client: Any, kwargs: Dict[str, Any], task: Optional[st # recovery chains (credential refresh, pool rotation, provider # fallback) see the same error they would on a plain call. if ( - _is_transient_transport_error(exc) + force_stream + or _is_transient_transport_error(exc) or _is_auth_error(exc) or _is_payment_error(exc) or _is_rate_limit_error(exc) @@ -7300,62 +7348,13 @@ def _aggregate_chat_stream( TimeoutError when *total_ceiling* seconds elapse before the stream finishes — phrased with "timed out" so existing timeout classification (``_is_timeout_error``) treats it exactly like a request timeout. + Accumulation is shared with the async mirror via + :class:`_ChatStreamAccumulator`. """ - started = time.monotonic() - content_parts: List[str] = [] - reasoning_parts: List[str] = [] - tool_calls_acc: Dict[int, Dict[str, Any]] = {} - finish_reason = None - usage = None - resp_id = "" - resp_model = model or "" + acc = _ChatStreamAccumulator(model=model, total_ceiling=total_ceiling) try: for chunk in chunks: - _notify_aux_progress() - if total_ceiling is not None and (time.monotonic() - started) >= total_ceiling: - raise TimeoutError( - f"Auxiliary streamed call timed out after {total_ceiling:.0f}s " - "total ceiling (stream still open but over budget)" - ) - resp_id = getattr(chunk, "id", None) or resp_id - resp_model = getattr(chunk, "model", None) or resp_model - chunk_usage = getattr(chunk, "usage", None) - if chunk_usage: - usage = chunk_usage - choices = getattr(chunk, "choices", None) or [] - if not choices: - continue - choice = choices[0] - finish_reason = getattr(choice, "finish_reason", None) or finish_reason - delta = getattr(choice, "delta", None) - if delta is None: - continue - piece = getattr(delta, "content", None) - if piece: - content_parts.append(piece) - # Thinking models stream reasoning deltas before any content — - # they count as forward progress (the tick above) and are kept - # out of .content, mirroring non-streaming responses where - # reasoning arrives in a separate field. - reasoning_piece = ( - getattr(delta, "reasoning", None) - or getattr(delta, "reasoning_content", None) - ) - if reasoning_piece and isinstance(reasoning_piece, str): - reasoning_parts.append(reasoning_piece) - for tc in (getattr(delta, "tool_calls", None) or []): - idx = getattr(tc, "index", 0) or 0 - acc = tool_calls_acc.setdefault( - idx, {"id": "", "name": "", "arguments": []} - ) - if getattr(tc, "id", None): - acc["id"] = tc.id - fn = getattr(tc, "function", None) - if fn is not None: - if getattr(fn, "name", None): - acc["name"] = fn.name - if getattr(fn, "arguments", None): - acc["arguments"].append(fn.arguments) + acc.feed(chunk) finally: close_fn = getattr(chunks, "close", None) if callable(close_fn): @@ -7363,38 +7362,157 @@ def _aggregate_chat_stream( close_fn() except Exception: pass + return acc.finish() - tool_calls = None - if tool_calls_acc: - tool_calls = [ - SimpleNamespace( - id=acc["id"], - type="function", - function=SimpleNamespace( - name=acc["name"], - arguments="".join(acc["arguments"]), - ), + +class _ChatStreamAccumulator: + """Shared per-chunk accumulation for sync and async stream aggregation. + + Mirrors :func:`_aggregate_chat_stream`'s chunk handling so the async + consumer below cannot drift from the sync one (same content/reasoning/ + tool-call delta reassembly, same "timed out" ceiling phrasing). + """ + + def __init__(self, model: str = "", total_ceiling: Optional[float] = None): + self._started = time.monotonic() + self._total_ceiling = total_ceiling + self.content_parts: List[str] = [] + self.reasoning_parts: List[str] = [] + self.tool_calls_acc: Dict[int, Dict[str, Any]] = {} + self.finish_reason = None + self.usage = None + self.resp_id = "" + self.resp_model = model or "" + + def feed(self, chunk: Any) -> None: + _notify_aux_progress() + if ( + self._total_ceiling is not None + and (time.monotonic() - self._started) >= self._total_ceiling + ): + raise TimeoutError( + f"Auxiliary streamed call timed out after {self._total_ceiling:.0f}s " + "total ceiling (stream still open but over budget)" ) - for _idx, acc in sorted(tool_calls_acc.items()) - ] + self.resp_id = getattr(chunk, "id", None) or self.resp_id + self.resp_model = getattr(chunk, "model", None) or self.resp_model + chunk_usage = getattr(chunk, "usage", None) + if chunk_usage: + self.usage = chunk_usage + choices = getattr(chunk, "choices", None) or [] + if not choices: + return + choice = choices[0] + self.finish_reason = getattr(choice, "finish_reason", None) or self.finish_reason + delta = getattr(choice, "delta", None) + if delta is None: + return + piece = getattr(delta, "content", None) + if piece: + self.content_parts.append(piece) + reasoning_piece = ( + getattr(delta, "reasoning", None) + or getattr(delta, "reasoning_content", None) + ) + if reasoning_piece and isinstance(reasoning_piece, str): + self.reasoning_parts.append(reasoning_piece) + for tc in (getattr(delta, "tool_calls", None) or []): + idx = getattr(tc, "index", 0) or 0 + acc = self.tool_calls_acc.setdefault( + idx, {"id": "", "name": "", "arguments": []} + ) + if getattr(tc, "id", None): + acc["id"] = tc.id + fn = getattr(tc, "function", None) + if fn is not None: + if getattr(fn, "name", None): + acc["name"] = fn.name + if getattr(fn, "arguments", None): + acc["arguments"].append(fn.arguments) - message = SimpleNamespace( - role="assistant", - content="".join(content_parts), - tool_calls=tool_calls, - reasoning="".join(reasoning_parts) or None, - ) - choice = SimpleNamespace( - index=0, - message=message, - finish_reason=finish_reason or "stop", - ) - return SimpleNamespace( - id=resp_id, - model=resp_model, - object="chat.completion", - choices=[choice], - usage=usage, + def finish(self) -> Any: + tool_calls = None + if self.tool_calls_acc: + tool_calls = [ + SimpleNamespace( + id=acc["id"], + type="function", + function=SimpleNamespace( + name=acc["name"], + arguments="".join(acc["arguments"]), + ), + ) + for _idx, acc in sorted(self.tool_calls_acc.items()) + ] + message = SimpleNamespace( + role="assistant", + content="".join(self.content_parts), + tool_calls=tool_calls, + reasoning="".join(self.reasoning_parts) or None, + ) + choice = SimpleNamespace( + index=0, + message=message, + finish_reason=self.finish_reason or "stop", + ) + return SimpleNamespace( + id=self.resp_id, + model=self.resp_model, + object="chat.completion", + choices=[choice], + usage=self.usage, + ) + + +async def _aggregate_chat_stream_async( + chunks: Any, + *, + model: str = "", + total_ceiling: Optional[float] = None, +) -> Any: + """Async mirror of :func:`_aggregate_chat_stream` (``async for`` consumer). + + The AsyncOpenAI stream contract is an async iterator — consuming it with + the sync helper raises. Same accumulation and ceiling semantics via + :class:`_ChatStreamAccumulator`. + """ + acc = _ChatStreamAccumulator(model=model, total_ceiling=total_ceiling) + try: + async for chunk in chunks: + acc.feed(chunk) + finally: + close_fn = getattr(chunks, "close", None) or getattr(chunks, "aclose", None) + if callable(close_fn): + try: + result = close_fn() + if inspect.isawaitable(result): + await result + except Exception: + pass + return acc.finish() + + +async def _acreate_with_stream( + client: Any, + kwargs: Dict[str, Any], + task: Optional[str] = None, +) -> Any: + """Async chat.completions.create() for stream-only providers. + + Sends ``stream=True`` and aggregates the async chunk stream into a + complete response (credit @kudi88, PR #60686 — async contract fixed to + ``async for`` and tool-call deltas preserved per sweeper review). + """ + total_ceiling = _aux_stream_total_ceiling(kwargs.get("timeout")) + stream_kwargs = dict(kwargs) + stream_kwargs["stream"] = True + stream_kwargs["stream_options"] = {"include_usage": True} + chunks = await client.chat.completions.create(**stream_kwargs) + # Defensive: shims may hand back a complete response despite stream=True. + if hasattr(chunks, "choices"): + return chunks + return await _aggregate_chat_stream_async( + chunks, model=str(kwargs.get("model") or ""), total_ceiling=total_ceiling, ) @@ -7598,7 +7716,13 @@ def call_llm( # for the transient retry every auxiliary task shares. (PR #16587) try: return _validate_llm_response( - _create_with_progress(client, kwargs, task), task, + _create_with_progress( + client, kwargs, task, + force_stream=_provider_requires_stream( + resolved_provider, _base_info or resolved_base_url, + ), + ), + task, provider=resolved_provider, base_url=_base_info) except Exception as transient_err: if not _is_transient_transport_error(transient_err): @@ -7631,7 +7755,13 @@ def call_llm( time.sleep(_backoff) try: return _validate_llm_response( - _create_with_progress(client, kwargs, task), task) + _create_with_progress( + client, kwargs, task, + force_stream=_provider_requires_stream( + resolved_provider, _base_info or resolved_base_url, + ), + ), + task) except Exception as retry_transient: if not _is_transient_transport_error(retry_transient): raise @@ -8184,9 +8314,25 @@ async def async_call_llm( # Retry ONCE on the same provider for a transient transport blip # before the except-chain escalates to fallback — see call_llm() # for the rationale. (PR #16587) + _force_stream_async = ( + _provider_requires_stream( + resolved_provider, _client_base or resolved_base_url, + ) + and not isinstance(client, ( + AsyncCodexAuxiliaryClient, + AsyncAnthropicAuxiliaryClient, + AsyncBedrockAuxiliaryClient, + )) + ) + + async def _acreate(_kwargs: Dict[str, Any]) -> Any: + if _force_stream_async: + return await _acreate_with_stream(client, _kwargs, task) + return await client.chat.completions.create(**_kwargs) + try: return _validate_llm_response( - await client.chat.completions.create(**kwargs), task, + await _acreate(kwargs), task, provider=resolved_provider, base_url=_client_base) except Exception as transient_err: if not _is_transient_transport_error(transient_err): @@ -8207,7 +8353,7 @@ async def async_call_llm( task or "call", transient_err, ) return _validate_llm_response( - await client.chat.completions.create(**kwargs), task) + await _acreate(kwargs), task) except Exception as first_err: if "temperature" in kwargs and _is_unsupported_temperature_error(first_err): retry_kwargs = dict(kwargs) diff --git a/contributors/emails/kudi3699@gmail.com b/contributors/emails/kudi3699@gmail.com new file mode 100644 index 000000000000..a4a578803bc5 --- /dev/null +++ b/contributors/emails/kudi3699@gmail.com @@ -0,0 +1 @@ +kudi88 diff --git a/hermes_cli/config.py b/hermes_cli/config.py index a82b328de5d9..0145244ccbc6 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1658,6 +1658,13 @@ DEFAULT_CONFIG = { # not a meaningful recovery, so an unretried blip silently loses the # call. "transient_retries": 2, + # Endpoints that reject NON-streaming chat requests outright (e.g. + # Tencent Copilot returns HTTP 400 "Non-stream chat request is + # currently not supported"). Auxiliary calls to a matching endpoint + # are sent with stream=True and aggregated client-side. Entries are + # case-insensitive substrings matched against the endpoint URL; + # copilot.tencent.com is always treated as stream-only. + "stream_only_base_urls": [], "vision": { "provider": "auto", # auto | openrouter | nous | codex | custom "model": "", # e.g. "google/gemini-2.5-flash", "gpt-4o" diff --git a/tests/agent/test_aux_progress_streaming.py b/tests/agent/test_aux_progress_streaming.py index 05e1ccb47d90..2eb0e5448e6f 100644 --- a/tests/agent/test_aux_progress_streaming.py +++ b/tests/agent/test_aux_progress_streaming.py @@ -10,15 +10,18 @@ hook, behavior is byte-for-byte the old non-streaming call. import threading import time from types import SimpleNamespace -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch import pytest from agent.auxiliary_client import ( + _acreate_with_stream, _aggregate_chat_stream, + _aggregate_chat_stream_async, _aux_stream_total_ceiling, _create_with_progress, _notify_aux_progress, + _provider_requires_stream, aux_progress_hook, ) from agent.conversation_compression import CompressionCommitFence @@ -271,3 +274,144 @@ class TestFenceProgress: with aux_progress_hook(fence.touch_progress): _notify_aux_progress() assert fence.seconds_since_progress() < 0.05 + + +# --------------------------------------------------------------------------- +# Stream-only providers (credit @kudi88, PR #60686) +# --------------------------------------------------------------------------- + +class TestProviderRequiresStream: + def test_tencent_copilot_is_stream_only(self): + assert _provider_requires_stream( + "custom", "https://copilot.tencent.com/v1" + ) is True + + def test_normal_endpoints_are_not(self): + assert _provider_requires_stream( + "openrouter", "https://openrouter.ai/api/v1" + ) is False + assert _provider_requires_stream("auto", None) is False + assert _provider_requires_stream("auto", "") is False + + def test_config_marker_matches_custom_endpoint(self): + with patch( + "hermes_cli.config.load_config", + return_value={"auxiliary": {"stream_only_base_urls": ["my-proxy.example.com"]}}, + ): + assert _provider_requires_stream( + "custom", "https://my-proxy.example.com/v1" + ) is True + assert _provider_requires_stream( + "custom", "https://other.example.com/v1" + ) is False + + def test_config_read_failure_fails_open_to_non_streaming(self): + with patch( + "hermes_cli.config.load_config", + side_effect=RuntimeError("config broken"), + ): + assert _provider_requires_stream( + "custom", "https://other.example.com/v1" + ) is False + + +class TestForceStream: + def test_force_stream_streams_without_a_hook(self): + chunks = [_chunk(content="hi", finish_reason="stop")] + client = _FakeClient(stream_chunks=chunks) + # NO aux_progress_hook installed — force_stream alone must stream. + result = _create_with_progress( + client, {"model": "m1", "messages": []}, force_stream=True, + ) + assert client.calls[0]["stream"] is True + assert result.choices[0].message.content == "hi" + + def test_force_stream_does_not_retry_nonstreaming_on_failure(self): + client = _FakeClient( + response=_COMPLETE, + stream_error=RuntimeError("HTTP 400 bad request"), + ) + with pytest.raises(RuntimeError, match="bad request"): + _create_with_progress( + client, {"model": "m1", "messages": []}, force_stream=True, + ) + # No silent non-streaming retry — the provider rejects those anyway. + assert len(client.calls) == 1 + + +class TestAsyncStreamAggregation: + @pytest.mark.asyncio + async def test_async_stream_is_consumed_with_async_for(self): + # The sweeper review of PR #60686 flagged that awaiting create() and + # then iterating synchronously raises — the async contract is + # ``async for``. Verify the async aggregator consumes a real async + # iterator and preserves tool-call deltas. + tc0 = SimpleNamespace( + index=0, id="call_9", + function=SimpleNamespace(name="lookup", arguments='{"q":'), + ) + tc1 = SimpleNamespace( + index=0, id=None, + function=SimpleNamespace(name=None, arguments='"x"}'), + ) + raw_chunks = [ + _chunk(content="part1 "), + _chunk(tool_calls=[tc0]), + _chunk(tool_calls=[tc1], content="part2", finish_reason="tool_calls"), + ] + + class _AsyncStream: + def __init__(self, items): + self._items = list(items) + self.closed = False + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._items: + raise StopAsyncIteration + return self._items.pop(0) + + async def close(self): + self.closed = True + + stream = _AsyncStream(raw_chunks) + result = await _aggregate_chat_stream_async(stream) + msg = result.choices[0].message + assert msg.content == "part1 part2" + assert msg.tool_calls[0].function.name == "lookup" + assert msg.tool_calls[0].function.arguments == '{"q":"x"}' + assert result.choices[0].finish_reason == "tool_calls" + assert stream.closed is True + + @pytest.mark.asyncio + async def test_acreate_with_stream_passes_stream_kwargs(self): + calls = [] + + class _AsyncStream: + def __init__(self, items): + self._items = list(items) + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._items: + raise StopAsyncIteration + return self._items.pop(0) + + class _AsyncClient: + def __init__(self): + completions = SimpleNamespace(create=self._create) + self.chat = SimpleNamespace(completions=completions) + + async def _create(self, **kwargs): + calls.append(kwargs) + return _AsyncStream([_chunk(content="ok", finish_reason="stop")]) + + result = await _acreate_with_stream( + _AsyncClient(), {"model": "m1", "messages": [], "timeout": 30}, + ) + assert calls[0]["stream"] is True + assert result.choices[0].message.content == "ok" diff --git a/website/docs/user-guide/configuration.md b/website/docs/user-guide/configuration.md index a5d20f6f8539..96ce47b7f6b2 100644 --- a/website/docs/user-guide/configuration.md +++ b/website/docs/user-guide/configuration.md @@ -985,6 +985,18 @@ If you do not want Hermes to auto-generate titles after the first exchange, set `auxiliary.title_generation.enabled: false`. Manual titles still work through `/title` and `hermes sessions rename`. +### Stream-only endpoints + +Some OpenAI-compatible endpoints reject non-streaming chat requests outright (e.g. Tencent Copilot returns HTTP 400 `"Non-stream chat request is currently not supported"`). Interactive chat already streams, but auxiliary tasks (title generation, compression, web extraction) use non-streaming calls and would fail on every attempt. Hermes always treats `copilot.tencent.com` as stream-only; for any other such endpoint, list a URL substring under `auxiliary.stream_only_base_urls`: + +```yaml +auxiliary: + stream_only_base_urls: + - "my-stream-only-proxy.example.com" +``` + +Matching auxiliary calls are sent with `stream=True` and the chunks (including tool-call deltas) are aggregated client-side — no behavior change for any other endpoint. + ### Video Tutorial