fix(codex): pass ConnectorRequestContext through HTTP transport adapter

Forward context to OpenAIConnector._handle_streaming_response so wire
capture and log correlation match the WebSocket path. Update unit
tests and _handle_streaming_response mocks to accept context kwargs.

Made-with: Cursor

Author: Mateusz

src/connectors/openai_codex/executor.py

@@ -141,7 +141,12 @@ async def initiate_streaming_request(
 
         # Default to HTTP/SSE
         return await self._connector._handle_streaming_response(  # type: ignore[misc]
-            url, payload, headers, session_id, "responses"
+            url,
+            payload,
+            headers,
+            session_id,
+            "responses",
+            context=context,
         )
 
     async def _initiate_websocket_streaming(

tests/unit/connectors/openai_codex/test_executor_non_streaming.py

@@ -90,7 +90,9 @@ async def second_iterator():
 
         captured_payloads: list[dict[str, object]] = []
 
-        async def streaming_side_effect(url, payload_dict, headers, session_id, *args):
+        async def streaming_side_effect(
+            url, payload_dict, headers, session_id, *args, **kwargs
+        ):
             captured_payloads.append(dict(payload_dict))
             if len(captured_payloads) == 1:
                 return first_handle

tests/unit/connectors/openai_codex/test_executor_streaming.py

@@ -815,7 +815,9 @@ async def second_iterator():
 
         captured_payloads: list[dict[str, object]] = []
 
-        async def streaming_side_effect(url, payload_dict, headers, session_id, *args):
+        async def streaming_side_effect(
+            url, payload_dict, headers, session_id, *args, **kwargs
+        ):
             captured_payloads.append(dict(payload_dict))
             if len(captured_payloads) == 1:
                 return first_handle

tests/unit/connectors/openai_codex/test_executor_websocket.py

@@ -1,15 +1,15 @@
 """Unit tests for ResponseExecutor WebSocket support."""
 
-from __future__ import annotations
-
-from typing import Any, cast
-from unittest.mock import AsyncMock, MagicMock, patch
-
-import pytest
-from src.connectors.contracts import ConnectorRequestContext
-from src.connectors.openai_codex.executor import _CodexTransportAdapter
-from src.core.common.exceptions import AuthenticationError
-from src.core.domain.responses import ProcessedResponse, StreamingResponseHandle
+from __future__ import annotations
+
+from typing import Any, cast
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from src.connectors.contracts import ConnectorRequestContext
+from src.connectors.openai_codex.executor import _CodexTransportAdapter
+from src.core.common.exceptions import AuthenticationError
+from src.core.domain.responses import ProcessedResponse, StreamingResponseHandle
 
 
 @pytest.mark.asyncio
@@ -62,13 +62,13 @@ async def mock_send_response_create(*args, **kwargs):
         async for chunk in handle.iterator:
             chunks.append(chunk)
 
-        # Verify chunks
-        assert len(chunks) == 2
-        # Websocket transport adapter yields ProcessedResponse objects directly
-        first_content = cast(dict[str, Any], chunks[0].content)
-        second_content = cast(dict[str, Any], chunks[1].content)
-        assert cast(dict[str, Any], first_content["message"])["content"] == "Hello"
-        assert cast(dict[str, Any], second_content["message"])["content"] == "World"
+        # Verify chunks
+        assert len(chunks) == 2
+        # Websocket transport adapter yields ProcessedResponse objects directly
+        first_content = cast(dict[str, Any], chunks[0].content)
+        second_content = cast(dict[str, Any], chunks[1].content)
+        assert cast(dict[str, Any], first_content["message"])["content"] == "Hello"
+        assert cast(dict[str, Any], second_content["message"])["content"] == "World"
 
     async def test_recreates_websocket_client_when_auth_token_changes(self) -> None:
         """Auth refresh retries must not reuse stale WebSocket credentials."""
@@ -117,17 +117,17 @@ async def test_initiate_websocket_streaming_no_auth(self) -> None:
 
         url = "https://chatgpt.com/backend-api/codex/responses"
         payload = {"model": "gpt-4", "input": []}
-        headers: dict[str, str] = {}  # No authorization header
+        headers: dict[str, str] = {}  # No authorization header
         session_id = "test_session"
 
         with pytest.raises(AuthenticationError, match="No API key"):
             await adapter.initiate_streaming_request(url, payload, headers, session_id)
 
-    async def test_http_fallback_when_websocket_disabled(self) -> None:
-        """Test fallback to HTTP/SSE when WebSocket is disabled."""
-        mock_connector = MagicMock()
-        mock_connector._handle_streaming_response = AsyncMock(
-            return_value=StreamingResponseHandle(
+    async def test_http_fallback_when_websocket_disabled(self) -> None:
+        """Test fallback to HTTP/SSE when WebSocket is disabled."""
+        mock_connector = MagicMock()
+        mock_connector._handle_streaming_response = AsyncMock(
+            return_value=StreamingResponseHandle(
                 iterator=AsyncMock(), headers={}, cancel_callback=AsyncMock()
             )
         )
@@ -143,54 +143,59 @@ async def test_http_fallback_when_websocket_disabled(self) -> None:
             url, payload, headers, session_id
         )
 
-        # Verify HTTP/SSE method was called
-        mock_connector._handle_streaming_response.assert_called_once_with(
-            url, payload, headers, session_id, "responses"
-        )
-        assert isinstance(handle, StreamingResponseHandle)
-
-    async def test_http_fallback_accepts_transport_metadata_kwargs(self) -> None:
-        """Transport adapter should accept the executor's keyword metadata contract."""
-        mock_connector = MagicMock()
-        mock_connector._handle_streaming_response = AsyncMock(
-            return_value=StreamingResponseHandle(
-                iterator=AsyncMock(), headers={}, cancel_callback=AsyncMock()
-            )
-        )
-
-        adapter = _CodexTransportAdapter(mock_connector, use_websocket=False)
-
-        url = "https://chatgpt.com/backend-api/codex/responses"
-        payload = {"model": "gpt-4", "input": []}
-        headers = {"Authorization": "Bearer test_key"}
-        session_id = "test_session"
-        request_context = ConnectorRequestContext(
-            request_id="req-1",
-            session_id="sess-1",
-            client_host="127.0.0.1",
-            extensions={},
-        )
-
-        handle = await adapter.initiate_streaming_request(
-            url,
-            payload,
-            headers,
-            session_id,
-            context=request_context,
-            backend="openai-codex",
-            model="gpt-4",
-            key_name="openai-codex",
-        )
-
-        mock_connector._handle_streaming_response.assert_called_once_with(
-            url, payload, headers, session_id, "responses"
-        )
-        assert isinstance(handle, StreamingResponseHandle)
-
-    async def test_cleanup_closes_websocket_client(self) -> None:
-        """Test cleanup properly disconnects WebSocket client."""
-        mock_connector = MagicMock()
-        adapter = _CodexTransportAdapter(mock_connector, use_websocket=True)
+        # Verify HTTP/SSE method was called with wire-capture context slot
+        mock_connector._handle_streaming_response.assert_called_once_with(
+            url, payload, headers, session_id, "responses", context=None
+        )
+        assert isinstance(handle, StreamingResponseHandle)
+
+    async def test_http_fallback_accepts_transport_metadata_kwargs(self) -> None:
+        """Transport adapter should accept the executor's keyword metadata contract."""
+        mock_connector = MagicMock()
+        mock_connector._handle_streaming_response = AsyncMock(
+            return_value=StreamingResponseHandle(
+                iterator=AsyncMock(), headers={}, cancel_callback=AsyncMock()
+            )
+        )
+
+        adapter = _CodexTransportAdapter(mock_connector, use_websocket=False)
+
+        url = "https://chatgpt.com/backend-api/codex/responses"
+        payload = {"model": "gpt-4", "input": []}
+        headers = {"Authorization": "Bearer test_key"}
+        session_id = "test_session"
+        request_context = ConnectorRequestContext(
+            request_id="req-1",
+            session_id="sess-1",
+            client_host="127.0.0.1",
+            extensions={},
+        )
+
+        handle = await adapter.initiate_streaming_request(
+            url,
+            payload,
+            headers,
+            session_id,
+            context=request_context,
+            backend="openai-codex",
+            model="gpt-4",
+            key_name="openai-codex",
+        )
+
+        mock_connector._handle_streaming_response.assert_called_once_with(
+            url,
+            payload,
+            headers,
+            session_id,
+            "responses",
+            context=request_context,
+        )
+        assert isinstance(handle, StreamingResponseHandle)
+
+    async def test_cleanup_closes_websocket_client(self) -> None:
+        """Test cleanup properly disconnects WebSocket client."""
+        mock_connector = MagicMock()
+        adapter = _CodexTransportAdapter(mock_connector, use_websocket=True)
 
         # Create mock WebSocket client
         mock_ws_client = AsyncMock()

tests/unit/connectors/test_openai_codex_codex_cli.py

@@ -622,6 +622,7 @@ async def streaming_side_effect(
             request_headers: dict[str, str],
             request_session_id: str,
             stream_format: str,
+            **kwargs: Any,
         ) -> StreamingResponseHandle:
             nonlocal call_count
             headers_seen.append(request_headers.get("Authorization"))
@@ -752,6 +753,7 @@ def handle_side_effect(
             request_headers: dict[str, str],
             request_session_id: str,
             stream_format: str,
+            **kwargs: Any,
         ) -> StreamingResponseHandle:
             headers_seen.append(request_headers.get("Authorization"))
             return stream_handles.pop(0)
@@ -849,6 +851,7 @@ async def streaming_side_effect(
             request_headers: dict[str, str],
             request_session_id: str,
             stream_format: str,
+            **kwargs: Any,
         ) -> StreamingResponseHandle:
             headers_seen.append(request_headers.get("Authorization"))
             raise HTTPException(status_code=401, detail="expired")
@@ -936,6 +939,7 @@ def handle_side_effect(
             request_headers: dict[str, str],
             request_session_id: str,
             stream_format: str,
+            **kwargs: Any,
         ) -> StreamingResponseHandle:
             return stream_handle