diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index aada15f51ed..87d3d6c4dbb 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -1944,6 +1944,35 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= request_client_holder["diag"] = _diag stream = request_client.chat.completions.create(**stream_kwargs) + # Some OpenAI-compatible adapters (for example copilot-acp) accept + # stream=True but still return a completed response object rather than + # an iterator of chunks. Treat that as "streaming unsupported" for the + # rest of this session instead of crashing on ``for chunk in stream`` + # with ``'types.SimpleNamespace' object is not iterable`` (#11732). + response_choices = getattr(stream, "choices", None) + if isinstance(response_choices, list) and response_choices: + logger.info( + "Streaming request returned a final response object instead of " + "an iterator; switching %s/%s to non-streaming for this session.", + agent.provider or "unknown", + agent.model or "unknown", + ) + agent._disable_streaming = True + message = getattr(response_choices[0], "message", None) + if message is not None: + reasoning_text = ( + getattr(message, "reasoning_content", None) + or getattr(message, "reasoning", None) + ) + if isinstance(reasoning_text, str) and reasoning_text: + _fire_first_delta() + agent._fire_reasoning_delta(reasoning_text) + content = getattr(message, "content", None) + if isinstance(content, str) and content: + _fire_first_delta() + agent._fire_stream_delta(content) + return stream + # Capture rate limit headers from the initial HTTP response. # The OpenAI SDK Stream object exposes the underlying httpx # response via .response before any chunks are consumed. diff --git a/tests/run_agent/test_streaming.py b/tests/run_agent/test_streaming.py index 134f789388c..2c4f8f8d013 100644 --- a/tests/run_agent/test_streaming.py +++ b/tests/run_agent/test_streaming.py @@ -624,6 +624,53 @@ class TestStreamingFallback: with pytest.raises(Exception, match="Connection reset by peer"): agent._interruptible_streaming_api_call({}) + @patch("run_agent.AIAgent._create_request_openai_client") + @patch("run_agent.AIAgent._close_request_openai_client") + def test_response_object_disables_streaming_and_returns_final_response( + self, mock_close, mock_create + ): + """Adapters that ignore stream=True should fall back cleanly.""" + from run_agent import AIAgent + + final_response = SimpleNamespace( + model="copilot-acp", + choices=[SimpleNamespace( + message=SimpleNamespace( + content="Hello from ACP", + tool_calls=None, + reasoning_content=None, + reasoning=None, + ), + finish_reason="stop", + )], + usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1, total_tokens=2), + ) + + mock_client = MagicMock() + mock_client.chat.completions.create.return_value = final_response + mock_create.return_value = mock_client + + agent = AIAgent( + model="claude-sonnet-4.6", + provider="copilot-acp", + api_key="test-key", + base_url="http://localhost:1234/v1", + quiet_mode=True, + skip_context_files=True, + skip_memory=True, + ) + agent.api_mode = "chat_completions" + agent._interrupt_requested = False + + deltas = [] + agent._stream_callback = lambda text: deltas.append(text) + + response = agent._interruptible_streaming_api_call({}) + + assert response is final_response + assert agent._disable_streaming is True + assert deltas == ["Hello from ACP"] + @patch("run_agent.AIAgent._create_request_openai_client") @patch("run_agent.AIAgent._close_request_openai_client") def test_stream_error_propagates_original(self, mock_close, mock_create):