fix: preserve refusal output and authentication in WebSocket helpers

src/openai/lib/azure.py

@@ -137,6 +137,13 @@ class BaseAzureClient(BaseClient[_HttpxClientT, _DefaultStreamT]):
     _azure_endpoint: httpx2.URL | None
     _azure_deployment: str | None
 
+    @property
+    @override
+    def auth_headers(self) -> dict[str, str]:
+        # WebSocket upgrades use this legacy accessor before prepared options.
+        # Reuse Azure's HTTP policy; prepared options own token refresh.
+        return self._auth_headers({"bearer_auth": True})
+
     @override
     def _build_request(
         self,

src/openai/lib/responses_websocket/README.md

@@ -71,7 +71,7 @@ the next response. It does not wait for a hypothetical future successor. Use
 `recv()` to observe that boundary when coordinating steering; accepted steering
 alone does not prove that a successor has started.
 
-For incremental text and tool-input snapshots, opt in with a separate helper
+For incremental text, refusal and tool-input snapshots, opt in with a separate helper
 fed by the events you already receive:
 
 ```python
@@ -139,7 +139,9 @@ for output in details["output"]:
 Logprobs accumulate with text deltas and a supplied `output_text.done.logprobs`
 replaces them, including empty or null values. Content/item/lifecycle replacements
 also replace their corresponding annotations and other metadata; text-only done
-events do not erase citations. Refusal, tool/MCP and unknown item/part fields are
+events do not erase citations. Refusal deltas extend `part["refusal"]` in detailed
+snapshots; `response.refusal.done` replaces that provisional value. Refusals do
+not become `snapshot().output_text`. Other tool/MCP and unknown item/part fields are
 retained as observed, with unset and null distinct. Unknown standalone events still
 pass through unchanged and are not accumulated. Annotation events enrich matching
 known items on the accumulator's lane; annotations received before any matching
@@ -247,6 +249,11 @@ retained input and output history as a new chain when needed, including tool
 and reasoning items rather than only displayed text. Request
 `reasoning.encrypted_content` when retaining reasoning-model output for replay.
 
+If you parsed a Responses result, retain its `output` objects for replay. The SDK
+keeps parsed text and tool arguments available for inspection while sending only
+their API fields when those typed objects are used in a request. Ordinary input
+dictionaries remain under your control and are sent as provided.
+
 ## Lanes, limits and connection lifetime
 
 Named stream IDs contain 1–256 ASCII letters, digits, underscores, hyphens or periods;

src/openai/lib/responses_websocket/_accumulator.py

@@ -49,6 +49,7 @@ class _Output:
     arguments: list[str] = field(default_factory=list[str])
     input: list[str] = field(default_factory=list[str])
     text: dict[str, list[str]] = field(default_factory=dict[str, list[str]])
+    refusals: dict[str, list[str]] = field(default_factory=dict[str, list[str]])
     data: dict[str, object] = field(default_factory=dict[str, object])
     content: dict[str, dict[str, object]] = field(default_factory=dict[str, dict[str, object]])
     annotations: dict[str, dict[str, object]] = field(default_factory=dict[str, dict[str, object]])
@@ -131,11 +132,14 @@ def detailed_snapshot(self) -> dict[str, Any]:
                 data["input"] = "".join(item.input)
             content: list[dict[str, object]] = []
             for pos in sorted(
-                item.content.keys() | item.text.keys() | item.annotations.keys(), key=lambda k: int(k, 16)
+                item.content.keys() | item.text.keys() | item.refusals.keys() | item.annotations.keys(),
+                key=lambda k: int(k, 16),
             ):
                 part = deepcopy(item.content.get(pos, {}))
                 if pos in item.text:
                     part["text"] = "".join(item.text[pos])
+                if pos in item.refusals:
+                    part["refusal"] = "".join(item.refusals[pos])
                 original = part.get("annotations")
                 if isinstance(original, list) or pos in item.annotations:
                     annotations = (
@@ -196,6 +200,8 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             "response.content_part.done",
             "response.output_text.delta",
             "response.output_text.done",
+            "response.refusal.delta",
+            "response.refusal.done",
             "response.output_text.annotation.added",
             "response.function_call_arguments.delta",
             "response.function_call_arguments.done",
@@ -293,13 +299,16 @@ def add_event(self, event: ResponsesServerEvent) -> None:
         value = ""
         if kind in {
             "response.output_text.delta",
+            "response.refusal.delta",
             "response.function_call_arguments.delta",
             "response.mcp_call_arguments.delta",
             "response.custom_tool_call_input.delta",
         }:
             value = _text_field(event, "delta")
         elif kind == "response.output_text.done":
             value = _text_field(event, "text")
+        elif kind == "response.refusal.done":
+            value = _text_field(event, "refusal")
         elif kind in {"response.function_call_arguments.done", "response.mcp_call_arguments.done"}:
             value = _text_field(event, "arguments")
         elif kind == "response.custom_tool_call_input.done":
@@ -314,9 +323,23 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             "response.content_part.done",
             "response.output_text.delta",
             "response.output_text.done",
+            "response.refusal.delta",
+            "response.refusal.done",
         } and (not isinstance(pos, int) or isinstance(pos, bool) or pos < 0):
             raise ValueError("WebSocket content_index must be a non-negative integer")
         item_id = _text_field(event, "item_id")
+        initial_refusal = ""
+        if kind in {"response.refusal.delta", "response.refusal.done"}:
+            previous = self._output.get(hex(index))
+            if previous is not None and (not previous.item_id or previous.item_id == item_id):
+                position = hex(pos)
+                part = previous.content.get(position, {})
+                if "type" in part and part["type"] != "refusal":
+                    return
+                # Preserve parts previously accepted as raw. Validate a seed
+                # only when a new refusal delta actually needs to extend it.
+                if kind == "response.refusal.delta" and position not in previous.refusals and "refusal" in part:
+                    initial_refusal = _text_field(part, "refusal")
         self._bound, self._stream_id = True, stream_id
         key = hex(index)
         item = self._output.get(key)
@@ -336,6 +359,8 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             "response.content_part.done",
             "response.output_text.delta",
             "response.output_text.done",
+            "response.refusal.delta",
+            "response.refusal.done",
         }:
             position = hex(pos)
             if kind == "response.output_text.delta":
@@ -351,6 +376,10 @@ def add_event(self, event: ResponsesServerEvent) -> None:
             elif kind == "response.output_text.done":
                 item.text[position] = [value]
                 item.content.setdefault(position, {}).update(_data(event, include={"logprobs"}))
+            elif kind == "response.refusal.delta":
+                item.refusals.setdefault(position, [initial_refusal]).append(value)
+            elif kind == "response.refusal.done":
+                item.refusals[position] = [value]
             else:
                 part = _field(event, "part")
                 item.content[position] = _data(part)
@@ -359,6 +388,7 @@ def add_event(self, event: ResponsesServerEvent) -> None:
                     item.text[position] = [value]
                 else:
                     item.text.pop(position, None)
+                item.refusals.pop(position, None)
         elif kind in {"response.function_call_arguments.delta", "response.mcp_call_arguments.delta"}:
             item.arguments.append(value)
         elif kind in {"response.function_call_arguments.done", "response.mcp_call_arguments.done"}:

src/openai/types/responses/parsed_response.py

@@ -50,6 +50,8 @@
 class ParsedResponseOutputText(ResponseOutputText, GenericModel, Generic[ContentType]):
     parsed: Optional[ContentType] = None
 
+    __api_exclude__ = {"parsed"}
+
 
 ParsedContent: TypeAlias = Annotated[
     Union[ParsedResponseOutputText[ContentType], ResponseOutputRefusal],
@@ -63,6 +65,21 @@ class ParsedResponseOutputMessage(ResponseOutputMessage, GenericModel, Generic[C
     else:
         content: List[ParsedContent]
 
+    @property
+    def __api_exclude__(self) -> dict[str, object]:
+        # The outer model dump flattens typed parts. Carry their exclusions to
+        # that dump, without interpreting ordinary dict keys as SDK metadata.
+        content = getattr(self, "content", None)
+        if not isinstance(content, list):
+            return {}
+        return {
+            "content": {
+                index: part.__api_exclude__
+                for index, part in enumerate(content)
+                if isinstance(part, ParsedResponseOutputText)
+            }
+        }
+
 
 class ParsedResponseFunctionToolCall(ResponseFunctionToolCall):
     parsed_arguments: object = None

tests/lib/responses/test_azure_responses_policy_regression.py

@@ -0,0 +1,102 @@
+"""Record both Azure upgrades completely, then compare SDK-key policy or caller override."""
+
+import json
+import asyncio
+from typing import Any
+from contextlib import AsyncExitStack
+
+import httpx2
+import pytest
+
+import openai
+from tests.lib.responses.test_websocket_session import script_server
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize("beta", [False, True])
+@pytest.mark.parametrize("caller_authorization", ["absent", "present", "omit"])
+async def test_azure_key_policy_survives_complete_responses_replacement(mode, beta, caller_authorization):
+    observed = []
+    restores = []
+
+    def peer(socket):
+        headers = socket.request.headers
+        observed.append(
+            {
+                "api-key": headers.get_all("api-key"),
+                "Authorization": headers.get_all("Authorization"),
+                "X-Caller": headers.get_all("X-Caller"),
+            }
+        )
+        if len(observed) == 1:
+            socket.send('{"type":"response.fixture","generation":1}')
+            assert json.loads(socket.recv(timeout=4)) == {"type": "response.fixture", "permit": "replacement"}
+            socket.close(1011, "fixture")
+        else:
+            socket.send('{"type":"response.fixture","generation":2}')
+            assert json.loads(socket.recv(timeout=4)) == {"type": "response.fixture", "received": 2}
+
+    def restore(event):
+        restores.append(event.close_code)
+
+    headers: dict[str, Any] = {"X-Caller": "retained"}
+    if caller_authorization == "present":
+        headers["aUtHoRiZaTiOn"] = "Bearer caller-selected-fixture"
+    elif caller_authorization == "omit":
+        headers["authorization"] = openai.omit
+    with script_server(peer, expected_connections=2) as url:
+        async with AsyncExitStack() as stack:
+            if mode == "sync":
+                cli = stack.enter_context(
+                    openai.AzureOpenAI(
+                        api_key="fake-azure-sdk-key",
+                        api_version="fixture",
+                        base_url=url,
+                        default_headers=headers,
+                        http_client=httpx2.Client(trust_env=False),
+                    )
+                )
+                resource = cli.beta.responses if beta else cli.responses
+                conn = stack.enter_context(resource.connect(on_reconnecting=restore, initial_delay=0, max_retries=1))
+                sync_iterator = iter(conn)
+
+                async def advance():
+                    return await asyncio.to_thread(lambda: next(sync_iterator))
+            else:
+                cli = await stack.enter_async_context(
+                    openai.AsyncAzureOpenAI(
+                        api_key="fake-azure-sdk-key",
+                        api_version="fixture",
+                        base_url=url,
+                        default_headers=headers,
+                        http_client=httpx2.AsyncClient(trust_env=False),
+                    )
+                )
+                resource = cli.beta.responses if beta else cli.responses
+                conn = await stack.enter_async_context(
+                    resource.connect(on_reconnecting=restore, initial_delay=0, max_retries=1)
+                )
+                async_iterator = aiter(conn)
+
+                async def advance():
+                    return await anext(async_iterator)
+
+            async def send(connection: Any, msg: dict[str, Any]):
+                if mode == "sync":
+                    await asyncio.to_thread(connection.send, msg)
+                else:
+                    await connection.send(msg)
+
+            assert (await asyncio.wait_for(advance(), 4)).generation == 1
+            await send(conn, {"type": "response.fixture", "permit": "replacement"})
+            assert (await asyncio.wait_for(advance(), 4)).generation == 2
+            await send(conn, {"type": "response.fixture", "received": 2})
+    assert restores == [1011]
+    # The peer completed both exchanges. A failure here cannot hide under the
+    # server helper's initial/expected-count cleanup assertion.
+    expected = {
+        "api-key": ["fake-azure-sdk-key"],
+        "Authorization": ["Bearer caller-selected-fixture"] if caller_authorization == "present" else [],
+        "X-Caller": ["retained"],
+    }
+    assert observed == [expected, expected]

tests/lib/responses/test_parsed_output_replay_profile.py

@@ -0,0 +1,102 @@
+"""Compare documented retained output versus user-converted inspection dictionaries."""
+
+import json
+import asyncio
+from copy import deepcopy
+from typing import Any
+from contextlib import AsyncExitStack
+
+import httpx2
+import pytest
+
+import openai
+from tests.lib.responses.test_websocket_session import script_server
+
+
+@pytest.mark.parametrize("mode", ["sync", "async"])
+@pytest.mark.parametrize("family", ["message", "function_call"])
+@pytest.mark.parametrize("representation", ["retained-output", "to-dict"])
+async def test_parsed_output_replay_observation_after_peer_cleanup(mode, family, representation):
+    item: dict[str, Any] = (
+        {
+            "id": "msg",
+            "type": "message",
+            "role": "assistant",
+            "phase": "commentary",
+            "status": "completed",
+            "content": [
+                {"type": "output_text", "text": "plain", "annotations": [], "logprobs": None},
+                # This is a server field on a different part type, not SDK parsing metadata.
+                {"type": "refusal", "refusal": "fixture refusal", "parsed": {"server_fixture": [None, False]}},
+            ],
+        }
+        if family == "message"
+        else {"type": "function_call", "call_id": "c1", "name": "never_execute", "arguments": '{"kind":"inert"}'}
+    )
+    item["future"] = {"nullable": [None, False, 0]}
+    raw = {
+        "id": "resp",
+        "object": "response",
+        "created_at": 1,
+        "model": "fixture-model",
+        "status": "completed",
+        "output": [deepcopy(item)],
+        "metadata": {"io": "fixture"},
+    }
+    expected = {
+        "type": "response.create",
+        "model": "fixture-model",
+        "input": [item],
+        "store": False,
+        "stream_id": "history",
+    }
+    seen = []
+
+    def http_peer(request):
+        assert request.url.path == "/v1/responses"
+        return httpx2.Response(200, json=raw)
+
+    def peer(socket):
+        seen.append(json.loads(socket.recv(timeout=4)))
+        socket.send(json.dumps({"type": "future.replay.confirm", "stream_id": "history"}))
+
+    with script_server(peer) as url:
+        async with AsyncExitStack() as stack:
+            if mode == "sync":
+                client = stack.enter_context(
+                    openai.OpenAI(
+                        api_key="fake",
+                        base_url="http://localhost/v1",
+                        websocket_base_url=url,
+                        http_client=httpx2.Client(transport=httpx2.MockTransport(http_peer)),
+                    )
+                )
+                conn = stack.enter_context(client.responses.connect(max_retries=0))
+            else:
+                client = await stack.enter_async_context(
+                    openai.AsyncOpenAI(
+                        api_key="fake",
+                        base_url="http://localhost/v1",
+                        websocket_base_url=url,
+                        http_client=httpx2.AsyncClient(transport=httpx2.MockTransport(http_peer)),
+                    )
+                )
+                conn = await stack.enter_async_context(client.responses.connect(max_retries=0))
+
+            async def call(obj, name, **kwargs):
+                if mode == "sync":
+                    return await asyncio.wait_for(asyncio.to_thread(getattr(obj, name), **kwargs), 4)
+                return await asyncio.wait_for(getattr(obj, name)(**kwargs), 4)
+
+            parsed = await call(client.responses, "parse", model="fixture-model", input="synthetic")
+            inspection = parsed.output[0].to_dict(exclude_unset=True)
+            history = parsed.output if representation == "retained-output" else [inspection]
+            await call(conn.response, "create", model="fixture-model", input=history, store=False, stream_id="history")
+            event = await call(conn, "recv")
+            assert event.to_dict(exclude_unset=True) == {"type": "future.replay.confirm", "stream_id": "history"}
+            assert parsed.output[0].to_dict(exclude_unset=True) == inspection
+    assert len(seen) == 1
+    if representation == "to-dict":
+        # Inspection dicts are caller-owned, including their parsed fields.
+        expected["input"] = [inspection]
+    assert seen[0] == expected