fix(llm): accept mapping Responses stream events

Signed-off-by: cgycorey <cgycorey@users.noreply.github.com>

agent/src/providers/llm.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 import logging
 import os
+from copy import copy
 from urllib.parse import urlparse
 import re
-from collections.abc import Mapping, Sequence
+from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
 from importlib import import_module
 from importlib.metadata import PackageNotFoundError, version
 from pathlib import Path
@@ -76,6 +77,164 @@ _AMBIENT_OPENAI_HEADER_ENV_VARS = (
 )
 
 
+class _ResponsesMappingEvent:
+    """Attribute view over a raw OpenAI Responses stream event mapping."""
+
+    def __init__(self, values: Mapping[str, Any]) -> None:
+        self._values = dict(values)
+
+    def __getattr__(self, name: str) -> Any:
+        try:
+            return self._values[name]
+        except KeyError as exc:
+            raise AttributeError(name) from exc
+
+    def model_dump(self, *, exclude_none: bool = False, **_: Any) -> dict[str, Any]:
+        """Expose the Pydantic method expected by LangChain's converter."""
+        if exclude_none:
+            return {
+                key: value for key, value in self._values.items() if value is not None
+            }
+        return dict(self._values)
+
+
+def _normalize_responses_stream_event(event: Any) -> Any:
+    """Adapt raw mapping events to the attribute shape LangChain expects.
+
+    Native OpenAI SDK streams yield typed response event objects. Some
+    OpenAI-compatible gateways yield the same wire events as plain dicts;
+    langchain-openai's Responses converter accesses ``event.type`` and a few
+    nested attributes directly. Preserve typed events unchanged and adapt only
+    the mapping form.
+    """
+    if not isinstance(event, Mapping):
+        return event
+
+    values = dict(event)
+    for key in ("annotation", "item"):
+        nested = values.get(key)
+        if isinstance(nested, Mapping):
+            values[key] = _ResponsesMappingEvent(nested)
+    return _ResponsesMappingEvent(values)
+
+
+class _ResponsesSyncStream:
+    """Proxy a sync Responses stream and normalize each yielded event."""
+
+    def __init__(self, stream: Any) -> None:
+        self._stream = stream
+        self._entered: Any = None
+
+    def __enter__(self) -> "_ResponsesSyncStream":
+        self._entered = self._stream.__enter__()
+        return self
+
+    def __exit__(self, *args: Any) -> Any:
+        return self._stream.__exit__(*args)
+
+    def __iter__(self) -> Iterator[Any]:
+        source = self._entered if self._entered is not None else self._stream
+        for event in source:
+            yield _normalize_responses_stream_event(event)
+
+    def parse(self, *args: Any, **kwargs: Any) -> "_ResponsesSyncStream":
+        """Keep raw-response ``parse()`` streams behind the same proxy."""
+        return _ResponsesSyncStream(self._stream.parse(*args, **kwargs))
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._stream, name)
+
+
+class _ResponsesAsyncStream:
+    """Proxy an async Responses stream and normalize each yielded event."""
+
+    def __init__(self, stream: Any) -> None:
+        self._stream = stream
+        self._entered: Any = None
+
+    async def __aenter__(self) -> "_ResponsesAsyncStream":
+        self._entered = await self._stream.__aenter__()
+        return self
+
+    async def __aexit__(self, *args: Any) -> Any:
+        return await self._stream.__aexit__(*args)
+
+    def __aiter__(self) -> AsyncIterator[Any]:
+        return self._iterate()
+
+    async def _iterate(self) -> AsyncIterator[Any]:
+        source = self._entered if self._entered is not None else self._stream
+        async for event in source:
+            yield _normalize_responses_stream_event(event)
+
+    def parse(self, *args: Any, **kwargs: Any) -> "_ResponsesAsyncStream":
+        """Keep async raw-response ``parse()`` streams behind the proxy."""
+        return _ResponsesAsyncStream(self._stream.parse(*args, **kwargs))
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._stream, name)
+
+
+class _ResponsesSyncResource:
+    """Proxy the sync Responses resource without changing non-stream calls."""
+
+    def __init__(self, resource: Any) -> None:
+        self._resource = resource
+
+    def create(self, *args: Any, **kwargs: Any) -> Any:
+        result = self._resource.create(*args, **kwargs)
+        if kwargs.get("stream"):
+            return _ResponsesSyncStream(result)
+        return result
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._resource, name)
+
+
+class _ResponsesAsyncResource:
+    """Proxy the async Responses resource without changing non-stream calls."""
+
+    def __init__(self, resource: Any) -> None:
+        self._resource = resource
+
+    async def create(self, *args: Any, **kwargs: Any) -> Any:
+        result = await self._resource.create(*args, **kwargs)
+        if kwargs.get("stream"):
+            return _ResponsesAsyncStream(result)
+        return result
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._resource, name)
+
+
+class _ResponsesSyncClient:
+    """Shallow client proxy used for one sync stream call."""
+
+    def __init__(self, client: Any) -> None:
+        self._client = client
+        self.responses = _ResponsesSyncResource(client.responses)
+        raw_response = getattr(client, "with_raw_response", None)
+        if raw_response is not None:
+            self.with_raw_response = _ResponsesSyncClient(raw_response)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._client, name)
+
+
+class _ResponsesAsyncClient:
+    """Shallow client proxy used for one async stream call."""
+
+    def __init__(self, client: Any) -> None:
+        self._client = client
+        self.responses = _ResponsesAsyncResource(client.responses)
+        raw_response = getattr(client, "with_raw_response", None)
+        if raw_response is not None:
+            self.with_raw_response = _ResponsesAsyncClient(raw_response)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._client, name)
+
+
 def _openai_custom_header_names(raw: str | None) -> tuple[str, ...]:
     """Return header names parsed by the OpenAI SDK environment format."""
     names: list[str] = []
@@ -361,6 +520,27 @@ if ChatOpenAI is not None:
                 self._capture(choice["message"], gen.message)
             return result
 
+        def _stream(self, *args: Any, **kwargs: Any) -> Iterator[Any]:
+            """Route Responses streams through the mapping-compatible adapter."""
+            if self._use_responses_api({**kwargs, **self.model_kwargs}):
+                cloned = copy(self)
+                cloned.root_client = _ResponsesSyncClient(self.root_client)
+                return super(ChatOpenAIWithReasoning, cloned)._stream(*args, **kwargs)
+            return super()._stream(*args, **kwargs)
+
+        async def _astream(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]:
+            """Async route matching ``_stream`` for Responses compatibility."""
+            if self._use_responses_api({**kwargs, **self.model_kwargs}):
+                cloned = copy(self)
+                cloned.root_async_client = _ResponsesAsyncClient(self.root_async_client)
+                async for chunk in super(ChatOpenAIWithReasoning, cloned)._astream(
+                    *args, **kwargs
+                ):
+                    yield chunk
+            else:
+                async for chunk in super()._astream(*args, **kwargs):
+                    yield chunk
+
         def _convert_chunk_to_generation_chunk(  # type: ignore[override]
             self,
             chunk: dict,
@@ -392,26 +572,27 @@ if ChatOpenAI is not None:
             is absent, breaking ReAct continuations after a tool call (#39).
             """
             payload = super()._get_request_payload(input_, stop=stop, **kwargs)
-            messages = super()._convert_input(input_).to_messages()
-            caps = self._capabilities()
-            for i, m in enumerate(payload["messages"]):
-                if m.get("role") != "assistant":
-                    continue
-                source_message = messages[i]
-                if caps.normalize_assistant_content and m.get("content") is None:
-                    m["content"] = ""
-                if caps.send_reasoning_content:
-                    m["reasoning_content"] = source_message.additional_kwargs.get(
-                        "reasoning_content", ""
-                    )
-                else:
-                    m.pop("reasoning_content", None)
-                if caps.gemini_thought_signatures:
-                    self._inject_tool_call_thought_signatures(
-                        m.get("tool_calls"), source_message
-                    )
-                else:
-                    self._strip_tool_call_extra_content(m.get("tool_calls"))
+            if "messages" in payload:
+                messages = super()._convert_input(input_).to_messages()
+                caps = self._capabilities()
+                for i, m in enumerate(payload["messages"]):
+                    if m.get("role") != "assistant":
+                        continue
+                    source_message = messages[i]
+                    if caps.normalize_assistant_content and m.get("content") is None:
+                        m["content"] = ""
+                    if caps.send_reasoning_content:
+                        m["reasoning_content"] = source_message.additional_kwargs.get(
+                            "reasoning_content", ""
+                        )
+                    else:
+                        m.pop("reasoning_content", None)
+                    if caps.gemini_thought_signatures:
+                        self._inject_tool_call_thought_signatures(
+                            m.get("tool_calls"), source_message
+                        )
+                    else:
+                        self._strip_tool_call_extra_content(m.get("tool_calls"))
 
             scoped_headers = self._provider_scoped_extra_headers()
             if scoped_headers:

agent/tests/test_responses_stream_compat.py

@@ -0,0 +1,138 @@
+"""Regression tests for OpenAI-compatible Responses event streams."""
+
+from __future__ import annotations
+
+import asyncio
+from contextlib import contextmanager
+from typing import Any, Iterator
+
+import pytest
+from langchain_core.messages import HumanMessage
+
+from src.providers.llm import ChatOpenAIWithReasoning
+
+
+class _MappingResponses:
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        self._events = events
+
+    @contextmanager
+    def create(self, **kwargs: Any) -> Iterator[Iterator[dict[str, Any]]]:
+        yield iter(self._events)
+
+
+class _MappingRootClient:
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        self.responses = _MappingResponses(events)
+
+
+class _MappingAsyncStream:
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        self._events = events
+
+    async def __aenter__(self) -> "_MappingAsyncStream":
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        return None
+
+    def __aiter__(self) -> Any:
+        return self._iterate()
+
+    async def _iterate(self) -> Any:
+        for event in self._events:
+            yield event
+
+
+class _MappingAsyncResponses:
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        self._events = events
+
+    async def create(self, **kwargs: Any) -> _MappingAsyncStream:
+        return _MappingAsyncStream(self._events)
+
+
+class _MappingAsyncRootClient:
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        self.responses = _MappingAsyncResponses(events)
+
+
+def _text_delta_event() -> dict[str, Any]:
+    return {
+        "type": "response.output_text.delta",
+        "delta": "hello",
+        "item_id": "msg_1",
+        "output_index": 0,
+        "content_index": 0,
+    }
+
+
+@pytest.mark.skipif(
+    ChatOpenAIWithReasoning is None,
+    reason="langchain-openai is not installed",
+)
+def test_responses_stream_accepts_mapping_events() -> None:
+    """OpenAI-compatible gateways may yield dicts instead of SDK event objects."""
+    events = [_text_delta_event()]
+    llm = ChatOpenAIWithReasoning(
+        model="gateway-reasoning-model",
+        api_key="sk-test",
+        use_responses_api=True,
+    )
+    llm.root_client = _MappingRootClient(events)
+    # Keep this test focused on stream-event compatibility rather than request
+    # serialization, which is exercised by the provider payload tests.
+    llm._get_request_payload = lambda *args, **kwargs: {
+        "model": "gateway-reasoning-model",
+        "input": [{"role": "user", "content": "hello"}],
+        "stream": True,
+    }
+
+    chunks = list(llm.stream("hello"))
+
+    assert "".join(chunk.text for chunk in chunks) == "hello"
+
+
+@pytest.mark.skipif(
+    ChatOpenAIWithReasoning is None,
+    reason="langchain-openai is not installed",
+)
+def test_async_responses_stream_accepts_mapping_events() -> None:
+    """The async Responses path must normalize gateway mappings as well."""
+    llm = ChatOpenAIWithReasoning(
+        model="gateway-reasoning-model",
+        api_key="sk-test",
+        use_responses_api=True,
+    )
+    llm.root_async_client = _MappingAsyncRootClient([_text_delta_event()])
+    llm._get_request_payload = lambda *args, **kwargs: {
+        "model": "gateway-reasoning-model",
+        "input": [{"role": "user", "content": "hello"}],
+        "stream": True,
+    }
+
+    async def collect() -> str:
+        chunks = [chunk async for chunk in llm.astream("hello")]
+        return "".join(chunk.text for chunk in chunks)
+
+    assert asyncio.run(collect()) == "hello"
+
+
+@pytest.mark.skipif(
+    ChatOpenAIWithReasoning is None,
+    reason="langchain-openai is not installed",
+)
+def test_responses_payload_does_not_assume_chat_messages() -> None:
+    """Responses requests use ``input`` and must bypass chat-only rewriting."""
+    llm = ChatOpenAIWithReasoning(
+        model="gateway-reasoning-model",
+        api_key="sk-test",
+        use_responses_api=True,
+        output_version="responses/v1",
+        reasoning={"effort": "high"},
+    )
+
+    payload = llm._get_request_payload([HumanMessage(content="hello")])
+
+    assert payload["input"]
+    assert payload["reasoning"] == {"effort": "high"}