fix(proxy): JSON-encode only plain hook chunks on Responses streams

A hook chunk whose whole tree is builtin JSON types streams as its named
event. Any other chunk keeps the base data frame, so nothing a custom
repr hides reaches the wire. Error event details encode the way the chat
error frame does.
This commit is contained in:
mateo-berri 2026-11-02 11:30:08 -08:00
parent 263e6aa4db
commit 6e82e7bd35
2 changed files with 89 additions and 976 deletions

View file

@ -1,13 +1,11 @@
import gc
import json
import re
import time
from collections import Counter, OrderedDict, defaultdict
from collections.abc import Iterable, Iterator, Mapping, Sequence
from collections.abc import Iterator, Mapping, Sequence
from contextlib import suppress
from enum import Enum
from http import HTTPStatus
from itertools import chain, groupby, takewhile, zip_longest
from itertools import accumulate, chain, groupby, islice, repeat, takewhile
from types import MappingProxyType
from typing import Final
@ -91,35 +89,8 @@ class _ErrorPayload(LiteLLMBaseModel):
return {"message": self.error} if isinstance(self.error, str) else self.error
_DROPPED: Final = object()
class _BaseForm(Enum):
PRINTED = "printed"
DUMPED = "dumped"
def _repr_owner(kind: type[object]) -> type[object]:
return next((owner for owner in kind.__mro__ if "__repr__" in owner.__dict__), object)
def _repr_prints_every_item(kind: type[object]) -> bool:
return kind in (OrderedDict, Counter) or _repr_owner(kind) in (dict, list, tuple, defaultdict)
def _format_prints_every_item(kind: type[object]) -> bool:
return kind.__str__ is object.__str__ and kind.__format__ is object.__format__ and _repr_prints_every_item(kind)
def _str_prints_every_item(kind: type[object]) -> bool:
return kind.__str__ is object.__str__ and _repr_owner(kind) in (
dict,
list,
tuple,
OrderedDict,
defaultdict,
Counter,
)
_JSON_LEAF_TYPES: Final = frozenset((str, int, float, bool, type(None)))
_MAX_PLAIN_JSON_DEPTH: Final = 100
def _is_dict(value: object) -> TypeIs[Mapping[object, object]]: # guard-ok: dict items as object
@ -130,201 +101,36 @@ def _is_sequence(value: object) -> TypeIs[Sequence[object]]: # guard-ok: items
return isinstance(value, (list, tuple))
def _is_proxy(value: object) -> TypeIs[Mapping[object, object]]: # guard-ok: mappingproxy cannot be subclassed
return isinstance(value, MappingProxyType)
def _is_defaultdict(value: object) -> TypeIs[defaultdict[object, object]]: # guard-ok: dict items as object
return isinstance(value, defaultdict)
def _proxied(proxy: Mapping[object, object]) -> object:
return gc.get_referents(proxy)[0]
def _repr_prints_its_fields(model: BaseModel) -> bool:
kind: Final = type(model)
return kind.__repr__ is BaseModel.__repr__ and kind.__repr_str__ is BaseModel.__repr_str__
def _plain_scalar(value: str | float) -> str | float:
if isinstance(value, str):
return str.__str__(value)
if isinstance(value, int):
return int.__int__(value)
return float.__float__(value)
def _repr_shows_its_value(value: str | float, printed: str) -> bool:
plain: Final = _plain_scalar(value)
def _is_plain_node(value: object) -> bool:
kind: Final = type(value)
if kind in _JSON_LEAF_TYPES or kind in (list, tuple):
return True
if isinstance(value, Enum):
return printed == f"<{type(value).__name__}.{value.name}: {plain!r}>"
return printed == repr(plain)
return isinstance(value, (str, int)) and kind.__repr__ is Enum.__repr__
return kind is dict and _is_dict(value) and all(type(key) is str for key in value)
def _printed_text(value: object) -> object:
printed: Final = repr(value)
if isinstance(value, (str, int, float)) and _repr_shows_its_value(value, printed):
return _plain_scalar(value)
return printed
def _children(value: object) -> tuple[object, ...]:
if _is_dict(value):
return tuple(value.values())
return tuple(value) if _is_sequence(value) else ()
def _dumped_text(value: object) -> object:
return _plain_scalar(value) if isinstance(value, (str, int, float)) else str(value)
def _next_level(level: tuple[object, ...], _: None) -> tuple[object, ...]:
return tuple(chain.from_iterable(_children(value) for value in level))
def _text(value: object, form: _BaseForm) -> object:
try:
return _printed_text(value) if form is _BaseForm.PRINTED else _dumped_text(value)
except Exception: # noqa: BLE001 # hook values run their own __repr__ and __str__, so a failure drops the value
return _DROPPED
def _json_key(key: object, form: _BaseForm) -> object:
return key if key is None or type(key) in (str, int, float, bool) else _text(key, form)
def _stored_sequence(value: Sequence[object]) -> Iterator[object]:
if isinstance(value, list):
return list[object].__iter__(value)
if isinstance(value, tuple):
return tuple[object, ...].__iter__(value)
return iter(value)
def _stored_items(value: Mapping[object, object]) -> Iterable[tuple[object, object]]:
return dict[object, object].items(value) if isinstance(value, dict) else value.items()
def _live_items(value: Mapping[object, object]) -> Iterator[tuple[object, object]]:
shown: Final = dict[int, object]()
while True:
with suppress(RuntimeError):
for key, item in _stored_items(value):
if id(key) not in shown:
shown[id(key)] = key
yield key, item
return
def _dumps_shown(kind: type[object], shown: Iterable[object], stored: Iterable[object]) -> bool:
return _str_prints_every_item(kind) or any(left is not right for left, right in zip_longest(shown, stored))
def _shown_sequence(value: object, form: _BaseForm) -> Iterable[object] | None:
if not _is_sequence(value):
return None
if form is _BaseForm.PRINTED:
return _stored_sequence(value) if _repr_prints_every_item(type(value)) else None
dumped: Final = tuple(value)
return dumped if _dumps_shown(type(value), dumped, _stored_sequence(value)) else None
def _shown_items(value: object, form: _BaseForm) -> Iterable[tuple[object, object]] | None:
if not _is_dict(value):
return None
if form is _BaseForm.DUMPED:
dumped: Final = tuple(value.items())
stored: Final = chain.from_iterable(_stored_items(value))
return dumped if _dumps_shown(type(value), chain.from_iterable(dumped), stored) else None
if not _repr_prints_every_item(type(value)):
return None
if isinstance(value, OrderedDict):
return tuple(OrderedDict[object, object].items(value))
if isinstance(value, Counter):
return Counter[object].most_common(value)
return _live_items(value)
def _proxy_items(proxy: Mapping[object, object], form: _BaseForm) -> Iterable[tuple[object, object]] | None:
mapping: Final = _proxied(proxy)
if form is _BaseForm.DUMPED and not _str_prints_every_item(type(mapping)):
return None
return _shown_items(mapping, _BaseForm.PRINTED)
def _expansion(value: object, form: _BaseForm) -> tuple[Iterable[tuple[object, object]], _BaseForm] | None:
if _is_proxy(value):
proxy_items: Final = _proxy_items(value, form)
return None if proxy_items is None else (proxy_items, _BaseForm.PRINTED)
shown_items: Final = _shown_items(value, form)
if shown_items is not None:
return shown_items, form
if form is _BaseForm.PRINTED and isinstance(value, BaseModel) and _repr_prints_its_fields(value):
return value.__repr_args__(), _BaseForm.PRINTED
return None
def _walked_ids(value: object) -> frozenset[int]:
return frozenset((id(value), id(_proxied(value)))) if _is_proxy(value) else frozenset((id(value),))
def _container_marker(value: object) -> object:
if isinstance(value, list):
return "[...]"
if isinstance(value, tuple):
return "(...)"
if isinstance(value, OrderedDict):
return "..."
if _is_defaultdict(value):
factory: Final = _text(value.default_factory, _BaseForm.PRINTED)
return factory if factory is _DROPPED else f"{type(value).__name__}({factory}, {{...}})"
return "{...}" if _is_dict(value) and not isinstance(value, Counter) else None
def _cycle_marker(value: object) -> object:
if not _is_proxy(value):
return _container_marker(value)
inner: Final = _container_marker(_proxied(value))
return inner if inner is None or inner is _DROPPED else f"mappingproxy({inner})"
def _cycle_text(value: object, form: _BaseForm) -> object:
marker: Final = _cycle_marker(value)
return _text(value, form) if marker is None else marker
def _json_safe(value: object, form: _BaseForm, path: frozenset[int]) -> object:
if value is None or type(value) in (str, int, float, bool):
return value
walked: Final = _walked_ids(value)
if not walked.isdisjoint(path):
return _cycle_text(value, form)
inner_path: Final = path | walked
try:
sequence: Final = _shown_sequence(value, form)
if sequence is not None:
shown_items: Final = [_json_safe(item, form, inner_path) for item in sequence]
return [item for item in shown_items if item is not _DROPPED]
expansion: Final = _expansion(value, form)
if expansion is None:
return _text(value, form)
items, inner = expansion
pairs: Final = [(_json_key(key, inner), _json_safe(item, inner, inner_path)) for key, item in items]
return {key: item for key, item in pairs if key is not _DROPPED and item is not _DROPPED}
except Exception: # noqa: BLE001 # hook values run their own printers and can nest past the recursion limit
return _text(value, form)
def _chunk_json(shown: object) -> str | None:
if isinstance(shown, str):
return shown
return None if shown is _DROPPED else json.dumps(shown)
def _is_plain_json(chunk: object) -> bool:
levels: Final = islice(accumulate(repeat(None), _next_level, initial=(chunk,)), _MAX_PLAIN_JSON_DEPTH)
return next((not level for level in levels if not level or not all(map(_is_plain_node, level))), False)
def responses_chunk_data(chunk: object) -> str | None:
if _is_proxy(chunk):
expandable: Final = _proxy_items(chunk, _BaseForm.DUMPED) is not None
return _chunk_json(_json_safe(chunk, _BaseForm.DUMPED, frozenset())) if expandable else None
if not _is_dict(chunk) or not _format_prints_every_item(type(chunk)):
kind: Final = type(chunk)
try:
return json.dumps(chunk, allow_nan=False) if kind is dict and _is_plain_json(chunk) else None
except (ValueError, RecursionError):
return None
return _chunk_json(_json_safe(chunk, _BaseForm.PRINTED, frozenset()))
def _dumped_detail(error: Mapping[str, object]) -> Mapping[str, object]:
path: Final = frozenset((id(error),))
shown: Final = [(key, _json_safe(value, _BaseForm.DUMPED, path)) for key, value in error.items()]
return {key: value for key, value in shown if value is not _DROPPED}
def sse_error_object(data: str | bytes) -> Mapping[str, object] | None:
@ -556,7 +362,7 @@ class ResponsesStreamErrorState:
def format_error(self, error: Mapping[str, object]) -> str:
fields: Final = responses_error_fields(error)
sequence_number: Final = self.sequence_number + 1
payload: Final = json.dumps(_error_event(fields, _dumped_detail(error), sequence_number))
payload: Final = json.dumps(_error_event(fields, error, sequence_number), default=str)
return f"event: error\ndata: {payload}\n\n"
def format_error_message(self, frame: str) -> str:

View file

@ -15,16 +15,12 @@ Pins covered:
from __future__ import annotations
import asyncio
import functools
import hashlib
import itertools
import json
import logging
from collections import Counter, OrderedDict, defaultdict
from collections.abc import AsyncIterator, Callable, ItemsView, Iterable, Iterator, KeysView, Mapping
from collections.abc import AsyncIterator, Callable, Iterable, Iterator, Mapping
from datetime import datetime, timezone
from enum import Enum
from http import HTTPStatus
from types import MappingProxyType
from typing import Final, Literal
from unittest.mock import AsyncMock, MagicMock
@ -34,8 +30,7 @@ import pytest
from fastapi import HTTPException, Response
from fastapi.responses import JSONResponse, StreamingResponse
from openai import APIError as OpenAIAPIError
from openai.types.responses import ResponseOutputMessage, ResponseOutputText
from pydantic import BaseModel, Field, PrivateAttr, model_serializer
from pydantic import BaseModel, model_serializer
import litellm
import litellm.proxy.proxy_server as ps
@ -172,8 +167,8 @@ async def test_responses_stream_dispatches_named_events_without_a_chat_done_sent
@pytest.mark.asyncio
@pytest.mark.parametrize(
"as_chunk",
[lambda event: event, lambda event: event.model_dump(), lambda event: MappingProxyType(event.model_dump())],
ids=["model", "dict", "mapping_proxy"],
[lambda event: event, lambda event: event.model_dump()],
ids=["model", "dict"],
)
async def test_responses_stream_sends_no_failure_event_after_an_upstream_error_event(
as_chunk: Callable[[ErrorEvent], BaseModel | Mapping[str, object]],
@ -204,17 +199,6 @@ async def test_responses_stream_sends_no_failure_event_after_an_upstream_error_e
assert json.loads(frames[-1].split("data: ", 1)[1])["message"] == "upstream failed"
class _MaskedEverywhereString(str):
def __repr__(self) -> str:
return "'***'"
def __str__(self) -> str:
return "***"
def __format__(self, format_spec: str) -> str:
return "***"
async def _responses_frames(chunks: Iterable[object]) -> tuple[str, ...]:
async def upstream() -> AsyncIterator[object]:
for chunk in chunks:
@ -230,13 +214,24 @@ async def _responses_frames(chunks: Iterable[object]) -> tuple[str, ...]:
)
_DUMPED_DELTA: Final = OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id="msg_1",
output_index=0,
content_index=0,
delta="hi",
sequence_number=3,
)
@pytest.mark.asyncio
async def test_responses_stream_never_writes_a_masked_type_value_on_its_event_line() -> None:
frames: Final = await _responses_frames(({"type": _MaskedEverywhereString("hook-canary"), "delta": "hi"},))
async def test_responses_stream_sends_a_hook_chunk_dumped_from_an_event_as_that_named_event() -> None:
frames: Final = await _responses_frames((_DUMPED_DELTA.model_dump(),))
assert len(frames) == 1
assert "hook-canary" not in frames[0]
assert json.loads(frames[0].split("data: ", 1)[1])["delta"] == "hi"
event_line, data_line = frames[0].strip().split("\n")
assert event_line == "event: response.output_text.delta"
assert json.loads(data_line.removeprefix("data: ")) == json.loads(_DUMPED_DELTA.model_dump_json())
@pytest.mark.asyncio
@ -271,60 +266,9 @@ async def test_responses_stream_names_a_model_chunk_from_the_json_it_sends() ->
assert json.loads(frames[0].removeprefix("data: ")) == {"sequence_number": 3}
class _HookOutputItem(BaseModel, frozen=True):
id: str
type: str
@pytest.mark.parametrize(
"item",
[
MappingProxyType({"id": "msg_nested", "type": "message"}),
_HookOutputItem(id="msg_nested", type="message"),
],
ids=["mapping_proxy", "model"],
)
def test_responses_sse_chunk_encodes_values_nested_in_a_hook_chunk_as_json_objects(
item: Mapping[str, object] | BaseModel,
) -> None:
frame: Final = _format_responses_sse_chunk(
chunk=MappingProxyType({"type": "response.output_item.added", "output_index": 0, "item": item})
)
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: "))["item"] == {"id": "msg_nested", "type": "message"}
def test_responses_error_event_encodes_nested_error_fields_as_json_objects() -> None:
frame: Final = ResponsesStreamErrorState().format_error(
{
"message": "overloaded",
"provider_specific_fields": MappingProxyType({"upstream": MappingProxyType({"type": "overloaded_error"})}),
}
)
payload: Final = json.loads(frame.split("data: ", 1)[1])
assert payload["error"]["provider_specific_fields"] == {"upstream": {"type": "overloaded_error"}}
_UPSTREAM_HEADERS: Final = httpx.Headers({"authorization": "Bearer hook-canary", "x-request-id": "req_1"})
def _responses_frame_payload(nested: object) -> Mapping[str, object]:
chunk_frame: Final = _format_responses_sse_chunk(
chunk=MappingProxyType({"type": "response.output_text.delta", "nested": nested})
)
error_frame: Final = ResponsesStreamErrorState().format_error({"message": "blocked", "nested": nested})
assert isinstance(chunk_frame, str)
chunk_payload: Final = json.loads(chunk_frame.removeprefix("data: "))
error_payload: Final = json.loads(error_frame.split("data: ", 1)[1])
assert chunk_payload["nested"] == error_payload["error"]["nested"]
return chunk_payload
class _HiddenFieldModel(BaseModel, frozen=True):
user: str
password: str = Field(repr=False)
class _MaskedReprString(str):
def __repr__(self) -> str:
return "'***'"
class _MaskedReprModel(BaseModel, frozen=True):
@ -334,502 +278,64 @@ class _MaskedReprModel(BaseModel, frozen=True):
return "_MaskedReprModel(token=[secure])"
class _MaskedReprStrModel(BaseModel, frozen=True):
token: str
def __repr_str__(self, join_str: str) -> str:
return "token=[secure]"
class _MaskedReprArgsModel(BaseModel, frozen=True):
token: str
def __repr_args__(self) -> Iterable[tuple[str | None, object]]:
return (("token", "[secure]"),)
class _MaskedStrModel(BaseModel, frozen=True):
token: str
def __str__(self) -> str:
return "_MaskedStrModel(token=[secure])"
class _MaskedReprString(str):
class _MaskedCount(int):
def __repr__(self) -> str:
return "_MaskedReprString([secure])"
return "_MaskedCount([secure])"
class _RigColor(Enum):
RED = "red"
def _self_referencing_chunk() -> Mapping[str, object]:
chunk: Final[dict[str, object]] = {"type": "response.output_text.delta"}
chunk["self"] = chunk # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return chunk
_REDACTED_HEADERS: Final = RedactedDict({"authorization": "Bearer hook-canary"})
_MASKED_IN_EVERY_FRAME: Final = (
_UPSTREAM_HEADERS,
_REDACTED_HEADERS,
MappingProxyType({"headers": _REDACTED_HEADERS}),
MappingProxyType(_REDACTED_HEADERS),
_HiddenFieldModel(user="rig", password="hook-canary"),
_MaskedReprStrModel(token="hook-canary"),
_MaskedReprArgsModel(token="hook-canary"),
)
_MASKED_IN_EVERY_FRAME_IDS: Final = (
"header_mapping",
"redacted_dict",
"proxy_over_dict_holding_redacted_dict",
"proxy_wrapping_redacted_dict",
"repr_false_field",
"masked_repr_str_model",
"masked_repr_args_model",
)
class _UnmaskingDict(dict[str, str]):
def items(self) -> ItemsView[str, str]: # pyright: ignore[reportIncompatibleMethodOverride] # rig hides storage
return {key: "hook-canary" for key in dict.keys(self)}.items()
class _UnmaskingList(list[str]):
def __iter__(self) -> Iterator[str]:
return iter(("hook-canary",))
class _MostCommonMaskedCounter(Counter[str]):
def most_common(self, n: int | None = None) -> list[tuple[str, int]]:
return [(key, count) for key, count in super().most_common(n) if key != "hook-canary"]
@pytest.mark.parametrize(
"nested",
[
*_MASKED_IN_EVERY_FRAME,
_MaskedReprModel(token="hook-canary"),
_MaskedReprString("hook-canary"),
_UnmaskingDict(token="enc:MASKED"),
_UnmaskingList(("enc:MASKED",)),
_MostCommonMaskedCounter(("shown", "hook-canary")),
],
ids=[
*_MASKED_IN_EVERY_FRAME_IDS,
"masked_repr_model",
"masked_repr_string",
"dict_items_differ_from_storage",
"list_iter_differs_from_storage",
"counter_most_common_masks",
],
)
def test_responses_sse_chunk_never_shows_what_a_nested_value_hides_from_its_repr(nested: object) -> None:
frame: Final = _format_responses_sse_chunk(
chunk=MappingProxyType({"type": "response.output_text.delta", "nested": nested})
)
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: "))["type"] == "response.output_text.delta"
assert "hook-canary" not in frame
class _OnceMaskedReprString(str):
def __init__(self, value: str) -> None:
super().__init__()
self.reprs: Final = itertools.count()
def __repr__(self) -> str:
return "'***'" if next(self.reprs) == 0 else super().__repr__()
@pytest.mark.parametrize(
"wrap,printed",
[
(lambda value: value, "'***'"),
(lambda value: MappingProxyType({"v": value}), {"v": "'***'"}),
],
ids=["dict_value", "nested_mapping_proxy"],
)
def test_responses_sse_chunk_prints_a_nested_string_once_the_way_base_printed_it(
wrap: Callable[[str], object], printed: object
) -> None:
frame: Final = _format_responses_sse_chunk(
chunk={"type": "response.output_text.delta", "v": wrap(_OnceMaskedReprString("hook-canary"))}
)
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: "))["v"] == printed
@pytest.mark.parametrize(
"nested",
[*_MASKED_IN_EVERY_FRAME, _MaskedStrModel(token="hook-canary")],
ids=[*_MASKED_IN_EVERY_FRAME_IDS, "masked_str_model"],
)
def test_responses_error_event_never_shows_what_a_nested_value_hides_from_its_string_form(nested: object) -> None:
frame: Final = ResponsesStreamErrorState().format_error(
{"message": "blocked", "provider_specific_fields": {"guardrail": "rig-guard", "detail": nested}}
)
payload: Final = json.loads(frame.split("data: ", 1)[1])
assert payload["error"]["provider_specific_fields"]["guardrail"] == "rig-guard"
assert "hook-canary" not in frame
class _OwnFormatDict(dict[str, str]):
def __format__(self, format_spec: str) -> str:
return json.dumps({"type": "response.output_text.delta", "token": "MASKED"})
class _MaskedStrDict(dict[str, str]):
def __str__(self) -> str:
return "{'type': 'response.output_text.delta', 'token': 'MASKED'}"
@pytest.mark.parametrize(
"chunk",
[
_OwnFormatDict(type="response.output_text.delta", token="hook-canary"),
MappingProxyType(_MaskedStrDict(type="response.output_text.delta", token="hook-canary")),
],
ids=["own_format", "proxy_over_masked_str"],
)
def test_responses_sse_chunk_sends_a_whole_chunk_as_the_text_its_printer_gave(chunk: object) -> None:
frame: Final = _format_responses_sse_chunk(chunk=chunk)
assert isinstance(frame, str)
assert "hook-canary" not in frame
class _RaisingRepr:
def __repr__(self) -> str:
raise RuntimeError("rig repr failure")
class _MaskedStrHolder(dict[str, object]):
def __str__(self) -> str:
return "{'type': 'response.output_text.delta', 'member': 'MASKED'}"
class _KeysRefusedDict(dict[str, str]):
def keys(self) -> KeysView[str]: # pyright: ignore[reportIncompatibleMethodOverride] # rig refuses its keys
raise RuntimeError("rig keys refused")
class _ItemRefusedDict(dict[str, str]):
def __getitem__(self, key: str) -> str:
raise RuntimeError("rig item refused")
class _PrintsOnceDict(dict[str, str]):
printed: bool = False
def __str__(self) -> str:
if self.printed:
raise RuntimeError("rig prints once")
self.printed = True
return "{'type': 'response.output_text.delta', 'token': 'MASKED'}"
_DELTA_STORAGE: Final = {"type": "response.output_text.delta", "token": "shown"}
class _AlternatingReprString(str):
def __init__(self, value: str) -> None:
super().__init__()
self.reprs: Final = itertools.count()
def __repr__(self) -> str:
return "'***'" if next(self.reprs) % 2 == 0 else super().__repr__()
class _ReprOnceString(str):
def __init__(self, value: str) -> None:
super().__init__()
self.reprs: Final = itertools.count()
def __repr__(self) -> str:
if next(self.reprs):
raise RuntimeError("rig repr works once")
return super().__repr__()
_PROXY_PLACEMENTS: Final[tuple[Callable[[str], object], ...]] = (
lambda value: {"type": "response.output_text.delta", "meta": MappingProxyType({"v": value})},
lambda value: MappingProxyType({"type": "response.output_text.delta", "v": value}),
lambda value: {"type": "response.output_text.delta", "meta": [MappingProxyType({"v": value})]},
)
@pytest.mark.parametrize("place", _PROXY_PLACEMENTS, ids=["nested_proxy", "whole_proxy", "proxy_in_list"])
def test_responses_sse_chunk_prints_each_proxy_item_once(place: Callable[[str], object]) -> None:
masked: Final = _format_responses_sse_chunk(chunk=place(_AlternatingReprString("hook-canary")))
shown: Final = _format_responses_sse_chunk(chunk=place(_ReprOnceString("hook-value")))
assert isinstance(masked, str)
assert isinstance(shown, str)
assert "'***'" in masked
assert "hook-canary" not in masked
assert "hook-value" in shown
def test_responses_error_event_prints_each_proxy_item_once() -> None:
frame: Final = ResponsesStreamErrorState().format_error(
{"message": "blocked", "provider_specific_fields": MappingProxyType({"v": _AlternatingReprString("hook-canary")})}
)
assert json.loads(frame.split("data: ", 1)[1])["error"]["provider_specific_fields"] == {"v": "'***'"}
def _dict_holding_itself(value: str) -> dict[str, object]:
held: Final[dict[str, object]] = {"v": value}
held["me"] = held # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return held
def _defaultdict_holding_itself(value: str) -> defaultdict[str, object]:
held: Final[defaultdict[str, object]] = defaultdict(list, v=value)
held["me"] = held # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return held
def _ordered_dict_holding_itself(value: str) -> OrderedDict[str, object]:
held: Final[OrderedDict[str, object]] = OrderedDict(v=value)
held["me"] = held # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return held
def _list_holding_itself(value: str) -> list[object]:
held: Final[list[object]] = [value]
held.append(held) # mutable-ok: a self-reference needs the list to exist before it can hold itself
return held
def _proxy_whose_mapping_holds_it(value: str) -> Mapping[str, object]:
held: Final[dict[str, object]] = {"type": "response.output_text.delta", "v": value}
held["me"] = MappingProxyType(held) # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return MappingProxyType(held)
_SELF_HOLDING_PLACEMENTS: Final[tuple[tuple[Callable[[str], object], Mapping[str, object]], ...]] = (
(
lambda value: {"type": "response.output_text.delta", "meta": MappingProxyType(_dict_holding_itself(value))},
{"type": "response.output_text.delta", "meta": {"v": "'***'", "me": "{...}"}},
),
(
lambda value: {"type": "response.output_text.delta", "meta": _dict_holding_itself(value)},
{"type": "response.output_text.delta", "meta": {"v": "'***'", "me": "{...}"}},
),
(
lambda value: {"type": "response.output_text.delta", "meta": _defaultdict_holding_itself(value)},
{"type": "response.output_text.delta", "meta": {"v": "'***'", "me": "defaultdict(<class 'list'>, {...})"}},
),
(
lambda value: {"type": "response.output_text.delta", "meta": _ordered_dict_holding_itself(value)},
{"type": "response.output_text.delta", "meta": {"v": "'***'", "me": "..."}},
),
(
lambda value: {"type": "response.output_text.delta", "meta": _list_holding_itself(value)},
{"type": "response.output_text.delta", "meta": ["'***'", "[...]"]},
),
(
_proxy_whose_mapping_holds_it,
{"type": "response.output_text.delta", "v": "'***'", "me": "mappingproxy({...})"},
),
)
@pytest.mark.parametrize(
("place", "expected"),
_SELF_HOLDING_PLACEMENTS,
ids=["proxy_over_dict", "dict", "defaultdict", "ordered_dict", "list", "proxy_held_by_its_mapping"],
)
def test_responses_sse_chunk_prints_a_container_that_holds_itself_as_its_repr_marker(
place: Callable[[str], object], expected: Mapping[str, object]
) -> None:
frame: Final = _format_responses_sse_chunk(chunk=place(_AlternatingReprString("hook-canary")))
assert isinstance(frame, str)
assert "hook-canary" not in frame
assert json.loads(frame.removeprefix("data: ")) == expected
class _ParentGrower:
def __init__(self, parent: dict[str, object]) -> None:
self.parent: Final = parent
def __repr__(self) -> str:
self.parent["added"] = 1 # mutable-ok: the hook value's repr grows the dict that holds it
return "_ParentGrower()"
def test_responses_sse_chunk_keeps_printing_a_dict_whose_item_repr_adds_a_key() -> None:
parent: Final[dict[str, object]] = {"v": _AlternatingReprString("hook-canary")}
parent["m"] = _ParentGrower(parent) # mutable-ok: the grower needs the dict it adds a key to
frame: Final = _format_responses_sse_chunk(chunk={"type": "response.output_text.delta", "meta": parent})
assert isinstance(frame, str)
assert "hook-canary" not in frame
assert json.loads(frame.removeprefix("data: "))["meta"] == {"v": "'***'", "m": "_ParentGrower()", "added": 1}
class _KeySwapper:
def __init__(self, parent: dict[str, object], removed: str, added: Mapping[str, object]) -> None:
self.parent: Final = parent
self.removed: Final = removed
self.added: Final = added
def __repr__(self) -> str:
del self.parent[self.removed] # mutable-ok: the hook value's repr resizes the dict that holds it
self.parent.update(self.added) # mutable-ok: the hook value's repr resizes the dict that holds it
return "_KeySwapper()"
@pytest.mark.parametrize(
("added", "later", "expected"),
[
({}, {"z1": "one", "z2": "two"}, {"a": "first", "d": "_KeySwapper()", "z1": "one", "z2": "two"}),
({"z": "zed"}, {}, {"a": "first", "d": "_KeySwapper()", "z": "zed"}),
],
ids=["removes_a_shown_key", "removes_a_shown_key_and_adds_one"],
)
def test_responses_sse_chunk_keeps_every_key_repr_prints_when_an_item_repr_resizes_its_dict(
added: Mapping[str, object], later: Mapping[str, object], expected: Mapping[str, object]
) -> None:
parent: Final[dict[str, object]] = {"a": "first"}
parent["d"] = _KeySwapper(parent, "a", added) # mutable-ok: the swapper needs the dict it resizes
parent.update(later) # mutable-ok: keys stored after the swapper
frame: Final = _format_responses_sse_chunk(chunk={"type": "response.output_text.delta", "meta": parent})
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: "))["meta"] == expected
class _StrFailsOnceDict(dict[str, str]):
failed: bool = False
def __str__(self) -> str:
if self.failed:
return dict.__repr__(self)
self.failed = True
raise RuntimeError("rig str fails on its first call")
def test_responses_sse_chunk_raises_when_a_whole_proxy_fails_its_only_print() -> None:
with pytest.raises(RuntimeError, match="rig str fails on its first call"):
_format_responses_sse_chunk(chunk=MappingProxyType(_StrFailsOnceDict(_DELTA_STORAGE)))
@pytest.mark.parametrize(
("make_underlying", "printed"),
[
(
lambda: _MaskedStrHolder(type="response.output_text.delta", member=_RaisingRepr()),
"{'type': 'response.output_text.delta', 'member': 'MASKED'}",
),
(lambda: _KeysRefusedDict(_DELTA_STORAGE), json.dumps(_DELTA_STORAGE)),
(lambda: _ItemRefusedDict(_DELTA_STORAGE), json.dumps(_DELTA_STORAGE)),
(
lambda: _PrintsOnceDict(type="response.output_text.delta", token="hook-canary"),
"{'type': 'response.output_text.delta', 'token': 'MASKED'}",
),
],
ids=["member_repr_raises", "keys_raise", "item_lookup_raises", "str_works_once"],
)
def test_responses_sse_chunk_sends_a_whole_proxy_as_its_mapping_printed_once(
make_underlying: Callable[[], Mapping[str, object]], printed: str
) -> None:
frame: Final = _format_responses_sse_chunk(chunk=MappingProxyType(make_underlying()))
assert frame == f"data: {printed}\n\n"
class _OwnFormatHolder(dict[str, str]):
def __format__(self, format_spec: str) -> str:
return "MASKED"
def _reordered() -> OrderedDict[str, int]:
ordered: Final = OrderedDict(first=1, second=2)
ordered.move_to_end("first")
return ordered
@pytest.mark.parametrize(
"nested",
[
_reordered(),
Counter(rare=1, common=3),
_OwnFormatHolder(token="shown"),
_MaskedStrDict(token="shown"),
],
ids=["reordered_ordered_dict", "counter", "own_format", "own_str"],
)
def test_responses_sse_chunk_sends_a_nested_mapping_with_the_items_and_order_its_repr_printed(
nested: Mapping[str, object],
) -> None:
frame: Final = _format_responses_sse_chunk(chunk={"type": "response.output_text.delta", "nested": nested})
assert isinstance(frame, str)
shown: Final = json.loads(frame.removeprefix("data: "))["nested"]
assert repr(shown) in repr(nested)
class _RigTuple(tuple[object, ...]):
pass
class _ItemsHideAliasDict(dict[str, str]):
__repr__ = dict.__repr__
def items(self) -> ItemsView[str, str]: # pyright: ignore[reportIncompatibleMethodOverride] # rig hides an entry
return {key: value for key, value in dict.items(self) if key != "h"}.items()
class _ItemsHideRevealingDict(_ItemsHideAliasDict):
def __repr__(self) -> str:
return dict.__repr__(self)
class _IterHideRevealingList(list[str]):
def __repr__(self) -> str:
return list.__repr__(self)
def __iter__(self) -> Iterator[str]:
return (item for item in list.__iter__(self) if item != "hook-canary")
class _RevealingReprModel(BaseModel, frozen=True):
a: int
_note: str = PrivateAttr(default="hook-canary")
def __repr__(self) -> str:
return f"_RevealingReprModel(note={self._note})"
@pytest.mark.parametrize(
"detail",
[
OrderedDict(member=_MaskedStrModel(token="hook-canary")),
_RigTuple((_MaskedStrModel(token="hook-canary"),)),
_HookOutputItem(id="msg_nested", type="message"),
_RevealingReprModel(a=1),
_ItemsHideAliasDict(k="shown", h="hook-canary"),
_ItemsHideRevealingDict(k="shown", h="hook-canary"),
_IterHideRevealingList(("shown", "hook-canary")),
{"type": _MaskedReprString("hook-canary"), "delta": "hi"},
{"type": "response.output_text.delta", "headers": _REDACTED_HEADERS},
RedactedDict({"type": "response.output_text.delta", "authorization": "Bearer hook-canary"}),
MappingProxyType({"type": "response.output_text.delta"}),
{"type": "response.output_text.delta", "item": _MaskedReprModel(token="hook-canary")},
{"type": "response.output_text.delta", "count": _MaskedCount(7)},
{"type": "response.output_text.delta", "color": _RigColor.RED},
{"type": "response.output_text.delta", ("a", "b"): 1},
{"type": "response.output_text.delta", "score": float("nan")},
_self_referencing_chunk(),
],
ids=[
"dict_subclass",
"tuple_subclass",
"model",
"masked_string",
"redacted_dict_value",
"redacted_dict_chunk",
"mapping_proxy_chunk",
"model_with_own_repr",
"items_hide_with_dict_repr_alias",
"items_hide_with_own_revealing_repr",
"iter_hides_with_own_revealing_repr",
"int_subclass_with_own_repr",
"enum_without_a_json_value",
"tuple_key",
"nan",
"holds_itself",
],
)
def test_responses_error_event_sends_a_detail_the_way_base_json_dumps_did(detail: object) -> None:
frame: Final = ResponsesStreamErrorState().format_error(
{"message": "blocked", "provider_specific_fields": {"detail": detail}}
)
def test_responses_sse_chunk_that_json_cannot_show_as_printed_keeps_the_base_frame(chunk: Mapping[object, object]) -> None:
assert _format_responses_sse_chunk(chunk=chunk) == _format_streaming_sse_chunk(chunk=chunk)
payload: Final = json.loads(frame.split("data: ", 1)[1])
assert payload["error"]["provider_specific_fields"]["detail"] == json.loads(json.dumps(detail, default=str))
def test_responses_error_event_encodes_its_detail_the_way_the_chat_error_frame_does() -> None:
error: Final = {
"message": "blocked",
"provider_specific_fields": {"guardrail": "rig-guard", "at": datetime(2026, 10, 30, tzinfo=timezone.utc)},
}
frame: Final = ResponsesStreamErrorState().format_error(error)
assert json.loads(frame.split("data: ", 1)[1])["error"] == json.loads(json.dumps(error, default=str))
@pytest.mark.parametrize("status_code", ["abc", 1.5, True, None], ids=["string", "float", "bool", "null"])
@ -870,205 +376,6 @@ def test_responses_failure_event_reads_a_numeric_status_code_the_way_base_did(st
assert failure.response.error["code"] == code
class _RigColor(Enum):
RED = 1
class _Score(float):
pass
class _MaskedCount(int):
def __repr__(self) -> str:
return "_MaskedCount([secure])"
_WHEN: Final = datetime(2026, 10, 30, 12, 0, tzinfo=timezone.utc)
@pytest.mark.parametrize(
("nested", "printed", "dumped"),
[
(_WHEN, repr(_WHEN), str(_WHEN)),
(_RigColor.RED, repr(_RigColor.RED), str(_RigColor.RED)),
(_MaskedCount(7), repr(_MaskedCount(7)), 7),
],
ids=["datetime", "enum", "number_subclass_with_its_own_repr"],
)
def test_responses_stream_frames_print_values_json_cannot_take_as_base_printed_them(
nested: object, printed: str, dumped: object
) -> None:
chunk_frame: Final = _format_responses_sse_chunk(
chunk=MappingProxyType({"type": "response.output_text.delta", "nested": nested})
)
error_frame: Final = ResponsesStreamErrorState().format_error({"message": "blocked", "nested": nested})
assert isinstance(chunk_frame, str)
assert json.loads(chunk_frame.removeprefix("data: "))["nested"] == printed
assert json.loads(error_frame.split("data: ", 1)[1])["error"]["nested"] == dumped
class _RigCode(str, Enum):
BLOCKED = "blocked_by_rig"
@pytest.mark.parametrize(
("nested", "plain"),
[(_RigCode.BLOCKED, "blocked_by_rig"), (HTTPStatus.BAD_REQUEST, 400), (_Score(0.5), 0.5)],
ids=["str_enum", "int_enum", "float_subclass"],
)
def test_responses_stream_frames_send_string_and_number_subclasses_as_their_plain_value(
nested: object, plain: object
) -> None:
assert _responses_frame_payload(nested)["nested"] == plain
_DUMPED_DELTA: Final = OutputTextDeltaEvent(
type=ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
item_id="msg_1",
output_index=0,
content_index=0,
delta="hi",
sequence_number=3,
)
@pytest.mark.parametrize("mapping", [dict, MappingProxyType], ids=["dict", "mapping_proxy"])
def test_responses_sse_chunk_a_hook_dumped_from_an_event_streams_as_that_event(
mapping: Callable[[Mapping[str, object]], Mapping[str, object]],
) -> None:
frame: Final = _format_responses_sse_chunk(chunk=mapping(_DUMPED_DELTA.model_dump()))
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: ")) == json.loads(_DUMPED_DELTA.model_dump_json())
_COMPLETED_WITH_SDK_OUTPUT: Final = ResponseCompletedEvent(
type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED,
sequence_number=4,
response=ResponsesAPIResponse(
id="resp_1",
object="response",
created_at=0,
model="gpt-x",
status="completed",
output=[
ResponseOutputMessage(
id="msg_1",
type="message",
role="assistant",
status="completed",
content=[ResponseOutputText(type="output_text", text="hi", annotations=[])],
)
],
),
)
def test_responses_sse_chunk_a_hook_copied_from_an_event_streams_as_that_event() -> None:
frame: Final = _format_responses_sse_chunk(chunk=dict(_COMPLETED_WITH_SDK_OUTPUT))
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: ")) == json.loads(_COMPLETED_WITH_SDK_OUTPUT.model_dump_json())
@pytest.mark.parametrize("mapping", [dict, MappingProxyType], ids=["dict", "mapping_proxy"])
def test_responses_stream_frames_turn_mapping_keys_json_cannot_take_into_strings(
mapping: Callable[[Mapping[object, object]], Mapping[object, object]],
) -> None:
assert _responses_frame_payload(mapping({("a", "b"): 1}))["nested"] == {repr(("a", "b")): 1}
def test_responses_sse_chunk_turns_top_level_keys_json_cannot_take_into_strings() -> None:
frame: Final = _format_responses_sse_chunk(
chunk=MappingProxyType({"type": "response.output_text.delta", ("a", "b"): "x"})
)
assert isinstance(frame, str)
assert json.loads(frame.removeprefix("data: ")) == {"type": "response.output_text.delta", repr(("a", "b")): "x"}
def _self_referencing_chunk() -> Mapping[str, object]:
chunk: Final[dict[str, object]] = {"type": "response.output_text.delta"}
chunk["self"] = chunk # mutable-ok: a self-reference needs the dict to exist before it can hold itself
return chunk
def test_responses_stream_frames_print_a_value_that_holds_itself_as_its_repr_marker() -> None:
nested: Final = _self_referencing_chunk()
assert _responses_frame_payload(nested)["nested"] == {"type": "response.output_text.delta", "self": "{...}"}
class _UnprintableValue:
def __repr__(self) -> str:
raise RuntimeError("no string form")
def test_responses_stream_frames_leave_out_only_the_nested_value_with_no_string_form() -> None:
chunk_frame: Final = _format_responses_sse_chunk(
chunk={"type": "response.output_text.delta", "meta": {"guardrail": "rig-guard", "bad": _UnprintableValue()}}
)
error_frame: Final = ResponsesStreamErrorState().format_error(
{
"message": "blocked",
"code": "400",
"type": "guardrail_error",
"provider_specific_fields": {"guardrail": "rig-guard", "bad": _UnprintableValue()},
}
)
assert isinstance(chunk_frame, str)
assert json.loads(chunk_frame.removeprefix("data: "))["meta"] == {"guardrail": "rig-guard"}
assert json.loads(error_frame.split("data: ", 1)[1])["error"] == {
"message": "blocked",
"code": "400",
"type": "guardrail_error",
"provider_specific_fields": {"guardrail": "rig-guard"},
}
def test_responses_error_event_keeps_a_detail_nested_past_the_recursion_limit() -> None:
nested: Final = functools.reduce(lambda inner, _: {"x": inner}, range(5000), {"leaf": 1})
frame: Final = ResponsesStreamErrorState().format_error(
{"message": "blocked", "provider_specific_fields": {"guardrail": "rig-guard", "detail": nested}}
)
payload: Final = json.loads(frame.split("data: ", 1)[1])
assert payload["error"]["provider_specific_fields"]["guardrail"] == "rig-guard"
assert "'leaf': 1" in frame
@pytest.mark.parametrize("mapping", [dict, MappingProxyType], ids=["dict", "mapping_proxy"])
def test_responses_stream_frames_keep_a_value_nested_700_levels_deep_as_json(
mapping: Callable[[Mapping[str, object]], Mapping[str, object]],
) -> None:
nested: Final = functools.reduce(lambda inner, _: mapping({"x": inner}), range(700), mapping({"leaf": 1}))
chunk_frame: Final = _format_responses_sse_chunk(chunk={"type": "response.output_text.delta", "nested": nested})
error_frame: Final = ResponsesStreamErrorState().format_error(
{"message": "blocked", "provider_specific_fields": {"detail": nested}}
)
assert isinstance(chunk_frame, str)
assert json.loads(chunk_frame.removeprefix("data: "))["nested"] == nested
assert json.loads(error_frame.split("data: ", 1)[1])["error"]["provider_specific_fields"]["detail"] == nested
@pytest.mark.parametrize(
"chunk",
[
_UPSTREAM_HEADERS,
RedactedDict({"type": "response.output_text.delta", "authorization": "Bearer hook-canary"}),
MappingProxyType(RedactedDict({"type": "response.output_text.delta", "authorization": "Bearer hook-canary"})),
],
ids=["header_mapping_chunk", "redacted_dict_chunk", "proxy_wrapping_redacted_dict_chunk"],
)
def test_responses_sse_chunk_that_is_not_a_plain_mapping_streams_as_the_plain_data_frame(
chunk: Mapping[str, object],
) -> None:
assert _format_responses_sse_chunk(chunk=chunk) == _format_streaming_sse_chunk(chunk=chunk)
@pytest.mark.asyncio
@pytest.mark.parametrize(
("guardrail_error", "error_object"),