From 156bbc28111dbfd0c5315c947d026ea939a1b584 Mon Sep 17 00:00:00 2001 From: markstuart-oai Date: Fri, 30 Oct 2026 02:40:02 +0000 Subject: [PATCH] fix: preserve refusal output and authentication in WebSocket helpers --- src/openai/lib/azure.py | 7 + src/openai/lib/responses_websocket/README.md | 11 +- .../lib/responses_websocket/_accumulator.py | 32 +++- src/openai/types/responses/parsed_response.py | 17 ++ .../test_azure_responses_policy_regression.py | 102 +++++++++++ .../test_parsed_output_replay_profile.py | 102 +++++++++++ .../test_refusal_normal_lifecycle.py | 167 ++++++++++++++++++ 7 files changed, 435 insertions(+), 3 deletions(-) create mode 100644 tests/lib/responses/test_azure_responses_policy_regression.py create mode 100644 tests/lib/responses/test_parsed_output_replay_profile.py create mode 100644 tests/lib/responses/test_refusal_normal_lifecycle.py diff --git a/src/openai/lib/azure.py b/src/openai/lib/azure.py index 82416b0ca7..4752505de5 100644 --- a/src/openai/lib/azure.py +++ b/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, diff --git a/src/openai/lib/responses_websocket/README.md b/src/openai/lib/responses_websocket/README.md index 2f32833f90..52833cca43 100644 --- a/src/openai/lib/responses_websocket/README.md +++ b/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; diff --git a/src/openai/lib/responses_websocket/_accumulator.py b/src/openai/lib/responses_websocket/_accumulator.py index cba9f5f82c..38e7f2e569 100644 --- a/src/openai/lib/responses_websocket/_accumulator.py +++ b/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,6 +299,7 @@ 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", @@ -300,6 +307,8 @@ def add_event(self, event: ResponsesServerEvent) -> None: 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"}: diff --git a/src/openai/types/responses/parsed_response.py b/src/openai/types/responses/parsed_response.py index 785578500c..6fbc52d339 100644 --- a/src/openai/types/responses/parsed_response.py +++ b/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 diff --git a/tests/lib/responses/test_azure_responses_policy_regression.py b/tests/lib/responses/test_azure_responses_policy_regression.py new file mode 100644 index 0000000000..8ffc37dcce --- /dev/null +++ b/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] diff --git a/tests/lib/responses/test_parsed_output_replay_profile.py b/tests/lib/responses/test_parsed_output_replay_profile.py new file mode 100644 index 0000000000..32a2e14f3a --- /dev/null +++ b/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 diff --git a/tests/lib/responses/test_refusal_normal_lifecycle.py b/tests/lib/responses/test_refusal_normal_lifecycle.py new file mode 100644 index 0000000000..c4d19c8905 --- /dev/null +++ b/tests/lib/responses/test_refusal_normal_lifecycle.py @@ -0,0 +1,167 @@ +"""Same enabled refusal bar with a full item lifecycle, before done/terminal.""" + +import json +from copy import deepcopy + +import pytest + +from openai.lib.responses_websocket import ResponsesWebSocketAccumulator +from tests.lib.responses.test_websocket_session import session_for, script_server, response_event + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_known_refusal_normal_created_item_part_deltas(mode): + frames = [ + response_event("created", output=None), + { + "type": "response.output_item.added", + "sequence_number": 1, + "output_index": 0, + "item": { + "type": "message", + "id": "msg_fixture", + "role": "assistant", + "status": "in_progress", + "content": [], + }, + }, + { + "type": "response.content_part.added", + "sequence_number": 2, + "output_index": 0, + "content_index": 0, + "item_id": "msg_fixture", + "part": {"type": "refusal", "refusal": ""}, + }, + { + "type": "response.refusal.delta", + "sequence_number": 3, + "output_index": 0, + "content_index": 0, + "item_id": "msg_fixture", + "delta": "Cannot ", + }, + { + "type": "response.refusal.delta", + "sequence_number": 4, + "output_index": 0, + "content_index": 0, + "item_id": "msg_fixture", + "delta": "do that", + }, + ] + + def peer(socket): + assert json.loads(socket.recv(timeout=4)) == {"type": "response.create", "input": "synthetic"} + for f in frames: + socket.send(json.dumps(f)) + # No done or terminal exists to repair the view under test. + + with script_server(peer) as url: + async with session_for(mode, url) as driver: + lane = driver.session.default + await driver.call(lane, "send", {"type": "response.create", "input": "synthetic"}) + acc = ResponsesWebSocketAccumulator() + observed = [] + for frame in frames[:3]: + event = await driver.call(lane, "recv") + original = deepcopy(event.to_dict(exclude_unset=True)) + assert original == frame + acc.add_event(event) + assert event.to_dict(exclude_unset=True) == original + observed.append((event, original)) + prior = acc.detailed_snapshot() + # Detailed item metadata excludes content, which is represented + # once as the separate indexed content rows. + assert prior["output"][0]["item"] == { + "type": "message", + "id": "msg_fixture", + "role": "assistant", + "status": "in_progress", + } + assert prior["output"][0]["content"][0]["part"] == {"type": "refusal", "refusal": ""} + for frame in frames[3:]: + event = await driver.call(lane, "recv") + original = deepcopy(event.to_dict(exclude_unset=True)) + assert original == frame + acc.add_event(event) + observed.append((event, original)) + actual = acc.detailed_snapshot() + assert all(event.to_dict(exclude_unset=True) == frame for event, frame in observed) + assert prior["output"][0]["content"][0]["part"]["refusal"] == "" + assert actual["output"][0]["content"][0]["part"]["refusal"] == "Cannot do that" + + +@pytest.mark.parametrize("mode", ["sync", "async"]) +async def test_refusal_sparse_seed_done_and_unknown_part_compatibility(mode: str) -> None: + pos = 2**63 + 31 + address = {"output_index": pos, "content_index": pos + 1, "item_id": "sparse"} + frames = [ + {"type": "response.refusal.delta", **address, "delta": "provisional", "sequence_number": 1}, + {"type": "response.refusal.done", **address, "refusal": "fixed", "sequence_number": 2}, + { + "type": "response.content_part.added", + **address, + "part": {"type": "refusal", "refusal": None}, + "sequence_number": 3, + }, + {"type": "response.refusal.delta", **address, "delta": "cannot use malformed seed", "sequence_number": 4}, + {"type": "response.refusal.done", **address, "refusal": "final correction", "sequence_number": 5}, + { + "type": "response.content_part.done", + **address, + "part": {"type": "future_part", "refusal": "caller-visible extra", "retained": [None, False]}, + "sequence_number": 6, + }, + {"type": "response.refusal.delta", **address, "delta": "must not annotate another type", "sequence_number": 7}, + ] + + def peer(socket): + assert json.loads(socket.recv(timeout=4)) == {"type": "response.create", "input": "synthetic"} + for frame in frames: + socket.send(json.dumps(frame)) + + with script_server(peer) as url: + async with session_for(mode, url) as driver: + lane = driver.session.default + await driver.call(lane, "send", {"type": "response.create", "input": "synthetic"}) + acc = ResponsesWebSocketAccumulator() + acc.add_event(await driver.call(lane, "recv")) + saved = acc.detailed_snapshot() + assert saved["output"] == [ + { + "output_index": pos, + "item": {"id": "sparse"}, + "content": [{"content_index": pos + 1, "part": {"refusal": "provisional"}}], + } + ] + assert acc.snapshot().output_text == "" + acc.add_event(await driver.call(lane, "recv")) + assert acc.detailed_snapshot()["output"][0]["content"][0]["part"] == {"refusal": "fixed"} + # Existing raw receipt accepted malformed parts before refusal deltas were supported. + part_event = await driver.call(lane, "recv") + before_raw = deepcopy(part_event.to_dict()) + acc.add_event(part_event) + before = acc.detailed_snapshot() + assert before["output"][0]["content"][0]["part"] == {"type": "refusal", "refusal": None} + delta_event = await driver.call(lane, "recv") + with pytest.raises(ValueError, match="refusal"): + acc.add_event(delta_event) + assert acc.detailed_snapshot() == before + assert part_event.to_dict() == before_raw + acc.add_event(await driver.call(lane, "recv")) + assert acc.detailed_snapshot()["output"][0]["content"][0]["part"] == { + "type": "refusal", + "refusal": "final correction", + } + acc.add_event(await driver.call(lane, "recv")) + future = acc.detailed_snapshot() + assert future["output"][0]["content"][0]["part"] == { + "type": "future_part", + "refusal": "caller-visible extra", + "retained": [None, False], + } + acc.add_event(await driver.call(lane, "recv")) + assert acc.detailed_snapshot() == future + assert saved["output"][0]["content"][0]["part"]["refusal"] == "provisional" + assert acc.snapshot().output_text == ""