fix: avoid forced Gemini CloudCode streaming

agent/gemini_cloudcode_adapter.py

@@ -450,7 +450,17 @@ def _make_stream_chunk(
     finish_reason: Optional[str] = None,
     reasoning: str = "",
 ) -> _GeminiStreamChunk:
-    delta_kwargs: Dict[str, Any] = {"role": "assistant"}
+    # Match the OpenAI SDK's delta shape: absent fields are present as None.
+    # run_agent's streaming path accesses delta.content and delta.tool_calls
+    # directly, so omitting them turns a valid text/reasoning-only Gemini chunk
+    # into AttributeError("SimpleNamespace has no attribute 'tool_calls'").
+    delta_kwargs: Dict[str, Any] = {
+        "role": "assistant",
+        "content": None,
+        "tool_calls": None,
+        "reasoning": None,
+        "reasoning_content": None,
+    }
     if content:
         delta_kwargs["content"] = content
     if tool_call_delta is not None:

run_agent.py

@@ -6966,6 +6966,18 @@ def _has_stream_consumers(self) -> bool:
             or getattr(self, "_stream_callback", None) is not None
         )
 
+    def _provider_requires_non_streaming(self) -> bool:
+        """Return True for providers where Hermes must avoid forced SSE.
+
+        Google Code Assist OAuth accepts the non-streaming generateContent
+        endpoint while the streaming endpoint can return account/model-level
+        429s even for the same request.  Do not force stream=True for this
+        provider just because the gateway has a stream consumer attached.
+        """
+        provider = str(getattr(self, "provider", "") or "").lower()
+        base_url = str(getattr(self, "base_url", "") or "").lower()
+        return provider == "google-gemini-cli" or base_url.startswith("cloudcode-pa://")
+
     def _interruptible_streaming_api_call(
         self, api_kwargs: dict, *, on_first_delta: callable = None
     ):
@@ -11582,6 +11594,8 @@ def _stop_spinner():
                     # session instead of re-failing every retry.
                     if getattr(self, "_disable_streaming", False):
                         _use_streaming = False
+                    elif self._provider_requires_non_streaming():
+                        _use_streaming = False
                     # CopilotACPClient communicates via subprocess stdio and
                     # returns a plain SimpleNamespace — not an iterable
                     # stream.  Mirror the ACP exclusion used for Responses

tests/agent/test_gemini_cloudcode.py

@@ -914,6 +914,23 @@ def test_finish_reason_switches_to_tool_calls_when_any_seen(self):
 
 
 class TestGeminiCloudCodeClient:
+    def test_text_stream_chunk_exposes_openai_delta_fields(self):
+        from agent.gemini_cloudcode_adapter import _translate_stream_event
+
+        chunks = _translate_stream_event(
+            {"response": {"candidates": [{
+                "content": {"parts": [{"text": "hello"}]},
+            }]}},
+            model="gemini-3.1-pro-preview",
+            tool_call_counter=[0],
+        )
+
+        delta = chunks[0].choices[0].delta
+        assert delta.content == "hello"
+        assert delta.tool_calls is None
+        assert delta.reasoning is None
+        assert delta.reasoning_content is None
+
     def test_client_exposes_openai_interface(self):
         from agent.gemini_cloudcode_adapter import GeminiCloudCodeClient
 

tests/run_agent/test_run_agent.py

@@ -4403,6 +4403,26 @@ def _make_tc_delta(index=0, tc_id=None, name=None, arguments=None):
 class TestStreamingApiCall:
     """Tests for _streaming_api_call — voice TTS streaming pipeline."""
 
+    def test_google_gemini_cli_requires_non_streaming_even_with_consumer(self, agent):
+        agent.provider = "google-gemini-cli"
+        agent.base_url = "cloudcode-pa://google"
+        agent.stream_delta_callback = MagicMock()
+
+        assert agent._has_stream_consumers() is True
+        assert agent._provider_requires_non_streaming() is True
+
+    def test_cloudcode_marker_base_url_requires_non_streaming(self, agent):
+        agent.provider = "custom"
+        agent.base_url = "cloudcode-pa://google"
+
+        assert agent._provider_requires_non_streaming() is True
+
+    def test_regular_openai_compatible_provider_can_stream(self, agent):
+        agent.provider = "openrouter"
+        agent.base_url = "https://openrouter.ai/api/v1"
+
+        assert agent._provider_requires_non_streaming() is False
+
     def test_content_assembly(self, agent):
         chunks = [
             _make_chunk(content="Hel"),