diff --git a/src/openai/_websocket_connection.py b/src/openai/_websocket_connection.py
index e74652a93c..320d18b313 100644
--- a/src/openai/_websocket_connection.py
+++ b/src/openai/_websocket_connection.py
@@ -1,22 +1,191 @@
 from __future__ import annotations
 
+import json
 import time
 import random
 import logging
 import threading
 import contextlib
-from typing import TYPE_CHECKING, Callable, Awaitable
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Mapping,
+    Callable,
+    Iterator,
+    Awaitable,
+    Generator,
+    AsyncIterator,
+    AsyncGenerator,
+    cast,
+)
 
 import anyio
 
 from ._types import Query, Headers
+from ._utils import is_given
+from ._models import FinalRequestOptions
+from ._exceptions import OpenAIError, WebSocketConnectionClosedError
 from ._send_queue import SendQueue, SendReceipt
+from ._base_client import _merge_mappings
 from .types.websocket_reconnection import ReconnectingEvent, ReconnectingOverrides, is_recoverable_close
 
 if TYPE_CHECKING:
+    import httpx2
     from websockets.sync.client import ClientConnection as WebSocketConnection
     from websockets.asyncio.client import ClientConnection as AsyncWebSocketConnection
 
+    from ._client import OpenAI, AsyncOpenAI
+
+
+def _websocket_request_options(
+    client: OpenAI | AsyncOpenAI, url: httpx2.URL, extra_query: Query, extra_headers: Headers
+) -> FinalRequestOptions:
+    try:
+        query_params = client._websocket_query_params
+    except AttributeError:
+        query_params = client._websocket_query_params_fallback
+    url = url.copy_with(params=query_params(url, extra_query))
+    url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme))
+    return FinalRequestOptions.construct(
+        method="get", url=str(url), headers=dict(extra_headers), security={"bearer_auth": True}
+    )
+
+
+def _websocket_connect_options(
+    client: OpenAI | AsyncOpenAI,
+    options: FinalRequestOptions,
+    connection_options: Mapping[str, Any],
+    default_headers: Headers,
+    logger: logging.Logger,
+) -> tuple[str, dict[str, str], dict[str, Any]]:
+    url = client._prepare_url(options.url).copy_merge_params(client.qs.stringify(cast(Any, options.params)))
+    url = url.copy_with(scheme={"http": "ws", "https": "wss"}.get(url.scheme, url.scheme))
+    headers = {
+        key.lower(): (key, value)
+        for header_set in (
+            client.auth_headers,
+            default_headers,
+            client.default_headers,
+            options.headers if is_given(options.headers) else {},
+        )
+        for key, value in header_set.items()
+    }
+    logger.debug("Connecting to WebSocket API")
+    if connection_options:
+        logger.debug("Custom WebSocket connection options provided")
+    websocket_options = dict(connection_options)
+    websocket_options.setdefault("max_size", None)
+    if websocket_options.get("max_size") is None:
+        websocket_options.setdefault("compression", None)
+    return str(url), _merge_mappings(dict(headers.values()), {}), websocket_options
+
+
+def _connect_websocket(
+    client: OpenAI,
+    prepare_url: Callable[[], httpx2.URL],
+    extra_query: Query,
+    extra_headers: Headers,
+    connection_options: Mapping[str, Any],
+    default_headers: Headers,
+    logger: logging.Logger,
+) -> WebSocketConnection:
+    try:
+        from websockets.sync.client import connect
+    except ImportError as exc:
+        raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc
+    options = client._prepare_options(_websocket_request_options(client, prepare_url(), extra_query, extra_headers))
+    url, headers, websocket_options = _websocket_connect_options(
+        client, options, connection_options, default_headers, logger
+    )
+    return connect(url, user_agent_header=None, additional_headers=headers, **websocket_options)
+
+
+async def _async_connect_websocket(
+    client: AsyncOpenAI,
+    prepare_url: Callable[[], httpx2.URL],
+    extra_query: Query,
+    extra_headers: Headers,
+    connection_options: Mapping[str, Any],
+    default_headers: Headers,
+    logger: logging.Logger,
+) -> AsyncWebSocketConnection:
+    try:
+        from .lib._websocket import _WebSocketConnect as connect
+    except ImportError as exc:
+        raise OpenAIError("You need to install `openai[realtime]` to use this method") from exc
+    options = await client._prepare_options(
+        _websocket_request_options(client, prepare_url(), extra_query, extra_headers)
+    )
+    url, headers, websocket_options = _websocket_connect_options(
+        client, options, connection_options, default_headers, logger
+    )
+    return await connect(url, user_agent_header=None, additional_headers=headers, **websocket_options)
+
+
+def _record_websocket_progress(
+    connection: WebSocketConnectionBase | AsyncWebSocketConnectionBase, event: object, *, raw: bool = False
+) -> None:
+    if raw:
+        if not connection._reconnect_attempt:
+            return
+        try:
+            event = json.loads(cast("str | bytes", event))
+        except (ValueError, RecursionError):
+            return
+    event_type = event.get("type") if isinstance(event, dict) else None if raw else getattr(event, "type", None)
+    if isinstance(event_type, str) and event_type and event_type != "error":
+        connection._reconnect_attempt = 0
+
+
+@contextlib.contextmanager
+def _websocket_entry_guard(
+    connection: WebSocketConnection, clear_connection: Callable[[], None]
+) -> Generator[None, None, None]:
+    try:
+        yield
+    except BaseException:
+        clear_connection()
+        try:
+            connection.close()
+        except BaseException:
+            # Preserve the exception that prevented successful entry.
+            pass
+        raise
+
+
+def _exit_websocket(connection: WebSocketConnectionBase | None, exc: BaseException | None) -> None:
+    if connection is not None:
+        try:
+            connection.close()
+        except Exception:
+            if exc is None:
+                raise
+
+
+async def _async_exit_websocket(connection: AsyncWebSocketConnectionBase | None, exc: BaseException | None) -> None:
+    if connection is not None:
+        try:
+            await connection.close()
+        except Exception:
+            if exc is None:
+                raise
+
+
+@contextlib.asynccontextmanager
+async def _async_websocket_entry_guard(
+    connection: AsyncWebSocketConnection, clear_connection: Callable[[], None]
+) -> AsyncGenerator[None, None]:
+    try:
+        yield
+    except BaseException:
+        clear_connection()
+        try:
+            await connection.close()
+        except BaseException:
+            # Preserve the exception that prevented successful entry.
+            pass
+        raise
+
 
 class WebSocketConnectionBase:
     def __init__(
@@ -52,6 +221,28 @@ def __init__(
         if interruptible_reconnect:
             self._close_event = threading.Event()
 
+    def _iter_events(self) -> Iterator[Any]:
+        from websockets.exceptions import ConnectionClosed, ConnectionClosedOK, ConnectionClosedError
+
+        while True:
+            try:
+                yield cast(Any, self).recv()
+            except ConnectionClosed as exc:
+                reconnect_exc = (
+                    ConnectionClosedError(exc.rcvd, exc.sent, exc.rcvd_then_sent)
+                    if isinstance(exc, ConnectionClosedOK)
+                    else exc
+                )
+                if not self._reconnect(reconnect_exc):
+                    unsent = self._send_queue.drain()
+                    if unsent:
+                        raise WebSocketConnectionClosedError(
+                            "WebSocket connection closed with unsent messages", unsent_messages=unsent
+                        ) from exc
+                    if isinstance(exc, ConnectionClosedOK):
+                        return
+                    raise
+
     def send_raw(self, data: bytes | str) -> None:
         self._send(data)
 
@@ -258,6 +449,28 @@ def __init__(
         self._send_queue = send_queue if send_queue is not None else SendQueue()
         self._logger = logger
 
+    async def _iter_events(self) -> AsyncIterator[Any]:
+        from websockets.exceptions import ConnectionClosed, ConnectionClosedOK, ConnectionClosedError
+
+        while True:
+            try:
+                yield await cast(Any, self).recv()
+            except ConnectionClosed as exc:
+                reconnect_exc = (
+                    ConnectionClosedError(exc.rcvd, exc.sent, exc.rcvd_then_sent)
+                    if isinstance(exc, ConnectionClosedOK)
+                    else exc
+                )
+                if not await self._reconnect(reconnect_exc):
+                    unsent = self._send_queue.drain()
+                    if unsent:
+                        raise WebSocketConnectionClosedError(
+                            "WebSocket connection closed with unsent messages", unsent_messages=unsent
+                        ) from exc
+                    if isinstance(exc, ConnectionClosedOK):
+                        return
+                    raise
+
     async def send_raw(self, data: bytes | str) -> None:
         await self._send(data)
 
diff --git a/tests/lib/test_websocket_connection_policy.py b/tests/lib/test_websocket_connection_policy.py
new file mode 100644
index 0000000000..2b6bb47e13
--- /dev/null
+++ b/tests/lib/test_websocket_connection_policy.py
@@ -0,0 +1,158 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+from unittest.mock import Mock, AsyncMock
+
+import httpx2
+import pytest
+
+from openai import OpenAI, AsyncOpenAI
+from openai._types import omit
+from openai._websocket_connection import (
+    _exit_websocket,
+    _connect_websocket,
+    _async_exit_websocket,
+    _websocket_entry_guard,
+    _async_connect_websocket,
+    _record_websocket_progress,
+    _async_websocket_entry_guard,
+)
+from openai.resources.responses.responses import ResponsesConnection, AsyncResponsesConnection
+
+LOG = logging.getLogger(__name__)
+
+
+@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize("key", [None, "additional_headers", "user_agent_header"])
+@pytest.mark.asyncio
+async def test_connect_keeps_prepared_policy_and_explicit_kwargs(
+    client: OpenAI, async_client: AsyncOpenAI, monkeypatch: pytest.MonkeyPatch, is_async: bool, key: str | None
+) -> None:
+    connector = AsyncMock() if is_async else Mock()
+    path = "openai.lib._websocket._WebSocketConnect" if is_async else "websockets.sync.client.connect"
+    monkeypatch.setattr(path, connector)
+
+    def prepare(options: Any) -> Any:
+        options.headers = {**options.headers, "X-Prepared": "kept"}
+        return options
+
+    sdk = async_client if is_async else client
+    monkeypatch.setattr(
+        sdk, "_prepare_options", AsyncMock(side_effect=prepare) if is_async else Mock(side_effect=prepare)
+    )
+    options = {} if key is None else {key: "caller"}
+
+    async def connect() -> None:
+        args: Any = (
+            lambda: httpx2.URL("https://example.test/v1/responses"),
+            {},
+            {"Authorization": omit},
+            options,
+            {},
+            LOG,
+        )
+        if is_async:
+            await _async_connect_websocket(async_client, *args)
+        else:
+            _connect_websocket(client, *args)
+
+    if key is not None:
+        with pytest.raises(TypeError, match="multiple values"):
+            await connect()
+        connector.assert_not_called()
+        return
+    await connect()
+    connector.assert_called_once()
+    assert connector.call_args.args == ("wss://example.test/v1/responses",)
+    kwargs = connector.call_args.kwargs
+    assert kwargs["additional_headers"]["X-Prepared"] == "kept"
+    assert "authorization" not in {header.lower() for header in kwargs["additional_headers"]}
+    assert kwargs["max_size"] is None and kwargs["compression"] is None
+    assert options == {}
+
+
+@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_iterator_keeps_dynamic_recv_and_generator_controls(
+    monkeypatch: pytest.MonkeyPatch, is_async: bool
+) -> None:
+    connection = AsyncResponsesConnection(Mock()) if is_async else ResponsesConnection(Mock())
+    monkeypatch.setattr(connection, "recv", AsyncMock(return_value="first") if is_async else Mock(return_value="first"))
+    events: Any = connection._iter_events()
+    assert (await anext(events) if is_async else next(events)) == "first"
+    monkeypatch.setattr(
+        connection, "recv", AsyncMock(return_value="second") if is_async else Mock(return_value="second")
+    )
+    assert (await anext(events) if is_async else next(events)) == "second"
+    error = ValueError("caller")
+    with pytest.raises(ValueError) as caught:
+        if is_async:
+            await events.athrow(error)
+        else:
+            events.throw(error)
+    assert caught.value is error
+    closing_events: Any = connection._iter_events()
+    assert (await anext(closing_events) if is_async else next(closing_events)) == "second"
+    if is_async:
+        await closing_events.aclose()
+        with pytest.raises(StopAsyncIteration):
+            await anext(closing_events)
+    else:
+        closing_events.close()
+        with pytest.raises(StopIteration):
+            next(closing_events)
+
+
+@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_entry_clears_before_cleanup_and_preserves_initiating_error(is_async: bool) -> None:
+    slot = [object()]
+    socket = Mock()
+
+    def close() -> None:
+        assert not slot
+        raise RuntimeError("secondary cleanup")
+
+    socket.close = AsyncMock(side_effect=close) if is_async else Mock(side_effect=close)
+    error = asyncio.CancelledError("initial flush") if is_async else KeyboardInterrupt("initial flush")
+    with pytest.raises(type(error)) as caught:
+        if is_async:
+            async with _async_websocket_entry_guard(socket, slot.clear):
+                raise error
+        else:
+            with _websocket_entry_guard(socket, slot.clear):
+                raise error
+    assert caught.value is error
+    socket.close.assert_called_once()
+
+
+@pytest.mark.parametrize("is_async", [False, True], ids=["sync", "async"])
+@pytest.mark.asyncio
+async def test_exit_preserves_body_error_over_ordinary_close_error(is_async: bool) -> None:
+    socket = Mock()
+    socket.close = AsyncMock(side_effect=RuntimeError("close")) if is_async else Mock(side_effect=RuntimeError("close"))
+    if is_async:
+        await _async_exit_websocket(socket, ValueError("body"))
+    else:
+        _exit_websocket(socket, ValueError("body"))
+    socket.close.assert_called_once()
+
+
+@pytest.mark.parametrize(
+    "event,raw,resets",
+    [
+        ({"type": "progress"}, False, True),
+        ({"type": "error"}, False, False),
+        (b'{"type":"progress"}', True, True),
+        (b'{"type":"error"}', True, False),
+        (b"invalid", True, False),
+        (b"[1]", True, False),
+    ],
+)
+def test_progress_requires_nonerror_event(event: Any, raw: bool, resets: bool) -> None:
+    connection = ResponsesConnection(Mock())
+    connection._reconnect_attempt = 3
+    _record_websocket_progress(connection, event, raw=raw)
+    assert connection._reconnect_attempt == (0 if resets else 3)