fix(responses): avoid replaying uncertain typed sends on reconnect (#4011)

A typed Responses WebSocket command could reach the server and then
raise locally. Reconnecting replayed that command, potentially creating
a duplicate response. The same problem occurred if sending a queued
command failed or was cancelled midway through a flush.

Queue only commands that have not been attempted. On a failed Responses
flush, discard the attempted item and retain the unsent tail in order
with its original byte limits. Keep typed sends on their existing
transport path so applications that customize `send_raw()` do not
unexpectedly intercept typed sends. This covers stable/beta and
sync/async; Realtime retains its existing queue policy.

Validation on this change:
- 294 affected Responses session, auth, preconnect, URL, queue and
Realtime recovery tests pass with both Pydantic v1 and v2.
- Extended the real WebSocket no-replay regression with a customized
raw-send policy: all four typed variants fail before the compatibility
fix and pass afterward; all four raw controls pass.
- Ruff, formatting and `git diff --check` pass.
- Local tests used the committed dependencies (`uv --frozen`): the local
uv is newer than the uv 0.12.5 pinned in CI. Hosted checks must still
pass on the published commit.

Author: markstuart-oai

src/openai/resources/beta/responses/responses.py

@@ -4342,11 +4342,7 @@ async def send(self, event: BetaResponsesClientEvent | BetaResponsesClientEventP
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            await self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        await self._connection.send(data)
 
     async def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -4454,7 +4450,7 @@ async def _send(data: str) -> None:
             await self._connection.send(data)
 
         try:
-            await self._send_queue.flush_async(_send)
+            await self._send_queue.flush_async(_send, requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 
@@ -4841,11 +4837,7 @@ def send(self, event: BetaResponsesClientEvent | BetaResponsesClientEventParam)
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        self._connection.send(data)
 
     def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -4947,7 +4939,7 @@ def _reconnect(self, exc: Exception) -> bool:
     def _flush_send_queue(self) -> None:
         """Send all queued messages over the current connection."""
         try:
-            self._send_queue.flush_sync(lambda data: self._connection.send(data))
+            self._send_queue.flush_sync(lambda data: self._connection.send(data), requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 

src/openai/resources/responses/responses.py

@@ -4193,11 +4193,7 @@ async def send(self, event: ResponsesClientEvent | ResponsesClientEventParam) ->
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            await self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        await self._connection.send(data)
 
     async def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -4305,7 +4301,7 @@ async def _send(data: str) -> None:
             await self._connection.send(data)
 
         try:
-            await self._send_queue.flush_async(_send)
+            await self._send_queue.flush_async(_send, requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 
@@ -4694,11 +4690,7 @@ def send(self, event: ResponsesClientEvent | ResponsesClientEventParam) -> None:
         if self._is_reconnecting:
             self._send_queue.enqueue(data)
             return
-        try:
-            self._connection.send(data)
-        except Exception:
-            self._send_queue.enqueue(data)
-            raise
+        self._connection.send(data)
 
     def send_raw(self, data: bytes | str) -> None:
         if self._is_reconnecting:
@@ -4810,7 +4802,7 @@ def _reconnect(self, exc: Exception) -> bool:
     def _flush_send_queue(self) -> None:
         """Send all queued messages over the current connection."""
         try:
-            self._send_queue.flush_sync(lambda data: self._connection.send(data))
+            self._send_queue.flush_sync(lambda data: self._connection.send(data), requeue_failed=False)
         except Exception:
             log.warning("Failed to flush send queue after reconnect")
 

tests/lib/responses/test_websocket_session.py

@@ -799,6 +799,93 @@ async def uncertain_async_send(data: Any) -> None:
                 await driver.call(lane, "send", {"type": "response.create"})
 
 
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize("beta", [False, True], ids=["stable", "beta"])
+@pytest.mark.parametrize("raw", [False, True], ids=["typed", "raw"])
+async def test_connection_does_not_replay_an_uncertain_create(
+    mode: str, beta: bool, raw: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    requests: list[dict[str, Any]] = []
+
+    def reject_raw(_data: bytes | str) -> None:
+        raise ValueError("Synthetic application policy: raw sends are disabled")
+
+    def reconnect(_event: ReconnectingEvent) -> ReconnectingOverrides:
+        return {"extra_headers": {"X-Recovery": "fresh"}}
+
+    def script(socket: ServerConnection) -> None:
+        assert socket.request is not None
+        if socket.request.headers.get("X-Recovery") == "fresh":
+            socket.send(json.dumps({"type": "response.future", "restored": True}))
+            while True:
+                received = json.loads(socket.recv(timeout=5))
+                requests.append(received)
+                if received == fresh_request:
+                    break
+        else:
+            requests.append(json.loads(socket.recv(timeout=5)))
+            socket.close(code=1011)
+
+    request = {"type": "response.create", "input": "synthetic uncertain request"}
+    fresh_request = {"type": "response.create", "input": "synthetic fresh work"}
+    with script_server(script, expected_connections=2) as url:
+        if mode == "sync":
+            with OpenAI(api_key="fake-key", base_url=url, http_client=httpx2.Client(trust_env=False)) as client:
+                resource = client.beta.responses if beta else client.responses
+                with resource.connect(on_reconnecting=reconnect, initial_delay=0, max_retries=1) as connection:
+                    if not raw:
+                        monkeypatch.setattr(connection, "send_raw", reject_raw)
+                    original_send = connection._connection.send
+
+                    def uncertain_send(data: Any) -> None:
+                        original_send(data)
+                        raise OSError("Synthetic failure after write")
+
+                    monkeypatch.setattr(connection._connection, "send", uncertain_send)
+                    with pytest.raises(OSError, match="Synthetic failure after write"):
+                        if raw:
+                            connection.send_raw(json.dumps(request))
+                        else:
+                            connection.send({"type": "response.create", "input": "synthetic uncertain request"})
+                    assert next(iter(connection)).to_dict()["restored"] is True
+                    if raw:
+                        connection.send_raw(json.dumps(fresh_request))
+                    else:
+                        connection.send({"type": "response.create", "input": "synthetic fresh work"})
+        else:
+            async with AsyncOpenAI(
+                api_key="fake-key", base_url=url, http_client=httpx2.AsyncClient(trust_env=False)
+            ) as async_client:
+                async_resource = async_client.beta.responses if beta else async_client.responses
+                async with async_resource.connect(
+                    on_reconnecting=reconnect, initial_delay=0, max_retries=1
+                ) as async_connection:
+                    if not raw:
+                        monkeypatch.setattr(async_connection, "send_raw", reject_raw)
+                    original_async_send = async_connection._connection.send
+
+                    async def uncertain_async_send(data: Any) -> None:
+                        await original_async_send(data)
+                        raise OSError("Synthetic failure after write")
+
+                    monkeypatch.setattr(async_connection._connection, "send", uncertain_async_send)
+                    with pytest.raises(OSError, match="Synthetic failure after write"):
+                        if raw:
+                            await async_connection.send_raw(json.dumps(request))
+                        else:
+                            await async_connection.send(
+                                {"type": "response.create", "input": "synthetic uncertain request"}
+                            )
+                    event = await asyncio.wait_for(anext(aiter(async_connection)), timeout=5)
+                    assert event.to_dict()["restored"] is True
+                    if raw:
+                        await async_connection.send_raw(json.dumps(fresh_request))
+                    else:
+                        await async_connection.send({"type": "response.create", "input": "synthetic fresh work"})
+
+    assert requests == [request, fresh_request]
+
+
 @pytest.mark.parametrize("mode", ["sync", "async"])
 async def test_early_clean_close_is_not_a_completed_response(mode: str) -> None:
     def script(socket: ServerConnection) -> None:
@@ -1353,8 +1440,9 @@ def check_then_reconnect(state: Any) -> None:
                 assert recovered.wait(5)
                 with session._router.condition:
                     assert session._router.condition.wait_for(
-                        lambda: session.connection._connection is not previous
-                        and not session.connection._is_reconnecting,
+                        lambda: (
+                            session.connection._connection is not previous and not session.connection._is_reconnecting
+                        ),
                         timeout=5,
                     )
                 if recover_before_check:

tests/test_send_queue_reconnect.py

@@ -108,15 +108,14 @@ def test_reconnect_retries_bounded_send_queue(
     q.enqueue("aaa")
     ws = MagicMock()
     attempts = 0
-    realtime = connection_type is RealtimeConnection
 
     def failing_send(data: str) -> None:
         nonlocal attempts
-        assert data == ("b" if realtime and attempts == 1 else "aaa")
+        assert data == ("b" if attempts == 1 else "aaa")
         if attempts == 0:
             q.enqueue("b")
         attempts += 1
-        if not realtime or attempts == 1:
+        if attempts == 1:
             with pytest.raises(WebSocketQueueFullError):
                 q.enqueue("c")
         raise RuntimeError("fake send failure")
@@ -132,10 +131,10 @@ def failing_send(data: str) -> None:
     )
     for expected_attempts in range(1, 4):
         assert connection._reconnect(RuntimeError("fake disconnect"))
-        assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4)
-        assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts)
+        assert q._bytes == (1 if expected_attempts == 1 else 0)
+        assert attempts == min(expected_attempts, 2)
     assert not connection._reconnect(RuntimeError("fake disconnect"))
-    assert attempts == (2 if realtime else 3)
+    assert attempts == 2
 
     # A healthy application event resets the budget; an upgrade alone does not.
     ws.recv.return_value = '{"type": "response.created"}'
@@ -144,7 +143,7 @@ def failing_send(data: str) -> None:
     sent: list[str] = []
     ws.send.side_effect = sent.append
     assert connection._reconnect(RuntimeError("fake disconnect"))
-    assert sent == ([] if realtime else ["aaa", "b"])
+    assert sent == []
     assert q._bytes == 0
     for _ in range(2):
         assert connection._reconnect(RuntimeError("fake disconnect"))
@@ -164,15 +163,14 @@ async def test_async_reconnect_retries_bounded_send_queue(
     q.enqueue("aaa")
     ws = MagicMock()
     attempts = 0
-    realtime = connection_type is AsyncRealtimeConnection
 
     async def failing_send(data: str) -> None:
         nonlocal attempts
-        assert data == ("b" if realtime and attempts == 1 else "aaa")
+        assert data == ("b" if attempts == 1 else "aaa")
         if attempts == 0:
             q.enqueue("b")
         attempts += 1
-        if not realtime or attempts == 1:
+        if attempts == 1:
             with pytest.raises(WebSocketQueueFullError):
                 q.enqueue("c")
         raise RuntimeError("fake send failure")
@@ -188,10 +186,10 @@ async def failing_send(data: str) -> None:
     )
     for expected_attempts in range(1, 4):
         assert await connection._reconnect(RuntimeError("fake disconnect"))
-        assert q._bytes == ((1 if expected_attempts == 1 else 0) if realtime else 4)
-        assert attempts == (min(expected_attempts, 2) if realtime else expected_attempts)
+        assert q._bytes == (1 if expected_attempts == 1 else 0)
+        assert attempts == min(expected_attempts, 2)
     assert not await connection._reconnect(RuntimeError("fake disconnect"))
-    assert attempts == (2 if realtime else 3)
+    assert attempts == 2
 
     # A healthy application event resets the budget; an upgrade alone does not.
     ws.recv = AsyncMock(return_value='{"type": "response.created"}')
@@ -200,15 +198,22 @@ async def failing_send(data: str) -> None:
     sent: list[str] = []
     ws.send.side_effect = sent.append
     assert await connection._reconnect(RuntimeError("fake disconnect"))
-    assert sent == ([] if realtime else ["aaa", "b"])
+    assert sent == []
     assert q._bytes == 0
     for _ in range(2):
         assert await connection._reconnect(RuntimeError("fake disconnect"))
     assert not await connection._reconnect(RuntimeError("retry budget exhausted"))
 
 
+@pytest.mark.parametrize(
+    "connection_type", [AsyncRealtimeConnection, AsyncResponsesConnection, AsyncBetaResponsesConnection]
+)
 @pytest.mark.asyncio
-async def test_cancelled_realtime_flush_does_not_replay_active_send() -> None:
+async def test_cancelled_flush_does_not_replay_active_send(
+    connection_type: type[AsyncRealtimeConnection]
+    | type[AsyncResponsesConnection]
+    | type[AsyncBetaResponsesConnection],
+) -> None:
     q = SendQueue(max_bytes=4)
     q.enqueue("é")
     q.enqueue("b")
@@ -221,7 +226,7 @@ async def suspended_send(data: str) -> None:
         await asyncio.Event().wait()
 
     socket.send = AsyncMock(side_effect=suspended_send)
-    connection = AsyncRealtimeConnection(socket, send_queue=q)
+    connection = connection_type(socket, send_queue=q)
     task = asyncio.create_task(connection._flush_send_queue())
     try:
         await asyncio.wait_for(active.wait(), 5)