From a7de0ea08ab3ae9faace6dcd9b4c965b28b968ae Mon Sep 17 00:00:00 2001
From: Solaris-star <820622658@qq.com>
Date: Mon, 20 Jul 2026 20:21:50 +0800
Subject: [PATCH] fix(openai): stream chat completions for long
 Anthropic-backed turns

Non-streaming chat.completions.create fails against Anthropic-backed
OpenAI-compatible endpoints with HTTP 400 "Streaming is required for
operations that may take longer than 10 minutes". Always stream and
reassemble content / tool_calls / usage into the classic response shape
consumed by existing deserializers.

Fixes #653
---
 .../node/agent/providers/openai_provider.py   |  5 +-
 .../test_openai_provider_stream_aggregate.py  | 82 +++++++++++++++++++
 2 files changed, 85 insertions(+), 2 deletions(-)
 create mode 100644 tests/unit/test_openai_provider_stream_aggregate.py

diff --git a/runtime/node/agent/providers/openai_provider.py b/runtime/node/agent/providers/openai_provider.py
index 809f6a00a7..51b361cb3d 100755
--- a/runtime/node/agent/providers/openai_provider.py
+++ b/runtime/node/agent/providers/openai_provider.py
@@ -6,6 +6,7 @@
 
 import binascii
 import os
+from types import SimpleNamespace
 from typing import Any, Dict, List, Optional, Union
 from urllib.parse import unquote_to_bytes
 
@@ -67,7 +68,7 @@ def call_model(
         
         if is_chat:
             request_payload = self._build_chat_payload(conversation, tool_specs, kwargs)
-            response = client.chat.completions.create(**request_payload)
+            response = self._chat_completion(client, request_payload)
             self._track_token_usage(response)
             self._append_chat_response_output(timeline, response)
             message = self._deserialize_chat_response(response)
@@ -83,7 +84,7 @@ def call_model(
             return ModelResponse(message=message, raw_response=response)
         except Exception as e:
             new_request_payload = self._build_chat_payload(conversation, tool_specs, kwargs)
-            response = client.chat.completions.create(**new_request_payload)
+            response = self._chat_completion(client, new_request_payload)
             self._track_token_usage(response)
             self._append_chat_response_output(timeline, response)
             message = self._deserialize_chat_response(response)
diff --git a/tests/unit/test_openai_provider_stream_aggregate.py b/tests/unit/test_openai_provider_stream_aggregate.py
new file mode 100644
index 0000000000..ba3d0309ea
--- /dev/null
+++ b/tests/unit/test_openai_provider_stream_aggregate.py
@@ -0,0 +1,82 @@
+"""Unit tests for OpenAIProvider streaming aggregation (#653)."""
+
+from types import SimpleNamespace
+
+from runtime.node.agent.providers.openai_provider import OpenAIProvider
+
+
+class _FakeStream:
+    def __init__(self, chunks):
+        self._chunks = chunks
+
+    def __iter__(self):
+        return iter(self._chunks)
+
+
+class _FakeClient:
+    def __init__(self, chunks):
+        self._chunks = chunks
+        self.last_kwargs = None
+
+        class _Completions:
+            def __init__(self, outer):
+                self._outer = outer
+
+            def create(self, **kwargs):
+                self._outer.last_kwargs = kwargs
+                return _FakeStream(self._outer._chunks)
+
+        class _Chat:
+            def __init__(self, outer):
+                self.completions = _Completions(outer)
+
+        self.chat = _Chat(self)
+
+
+def _delta_chunk(content=None, tool_calls=None, usage=None, id_="chatcmpl-1", model="gpt-test"):
+    delta = SimpleNamespace(content=content, tool_calls=tool_calls)
+    choice = SimpleNamespace(delta=delta, index=0)
+    return SimpleNamespace(id=id_, model=model, choices=[choice], usage=usage)
+
+
+def test_chat_completion_forces_stream_and_aggregates_content():
+    chunks = [
+        _delta_chunk(content="Hello"),
+        _delta_chunk(content=" world", usage=SimpleNamespace(prompt_tokens=1, completion_tokens=2, total_tokens=3)),
+    ]
+    client = _FakeClient(chunks)
+    provider = OpenAIProvider(model_name="gpt-test", params={})
+
+    response = provider._chat_completion(client, {"model": "gpt-test", "messages": []})
+
+    assert client.last_kwargs["stream"] is True
+    assert client.last_kwargs["stream_options"] == {"include_usage": True}
+    assert response.choices[0].message.content == "Hello world"
+    assert response.usage.total_tokens == 3
+
+
+def test_chat_completion_aggregates_tool_call_deltas():
+    tc0_a = SimpleNamespace(
+        index=0,
+        id="call_1",
+        type="function",
+        function=SimpleNamespace(name="search", arguments='{"q":'),
+    )
+    tc0_b = SimpleNamespace(
+        index=0,
+        id=None,
+        type=None,
+        function=SimpleNamespace(name=None, arguments='"ai"}'),
+    )
+    chunks = [
+        _delta_chunk(tool_calls=[tc0_a]),
+        _delta_chunk(tool_calls=[tc0_b]),
+    ]
+    client = _FakeClient(chunks)
+    provider = OpenAIProvider(model_name="gpt-test", params={})
+
+    response = provider._chat_completion(client, {"model": "gpt-test", "messages": []})
+    tc = response.choices[0].message.tool_calls[0]
+    assert tc.id == "call_1"
+    assert tc.function.name == "search"
+    assert tc.function.arguments == '{"q":"ai"}'