From 4180859208d00e16090ea1cc54a4552d5763f04e Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Fri, 30 Oct 2026 12:20:16 -0700 Subject: [PATCH] fix(proxy): print values in Responses stream frames the way each base path did --- .../common_utils/responses_stream_errors.py | 109 +++++++--- .../proxy_server/test_streaming_helpers.py | 188 +++++++++++++++--- 2 files changed, 238 insertions(+), 59 deletions(-) diff --git a/litellm/proxy/common_utils/responses_stream_errors.py b/litellm/proxy/common_utils/responses_stream_errors.py index 88ef6c25422..0f9df6b6985 100644 --- a/litellm/proxy/common_utils/responses_stream_errors.py +++ b/litellm/proxy/common_utils/responses_stream_errors.py @@ -1,6 +1,7 @@ import json import time from collections.abc import Iterable, Mapping, Sequence +from enum import Enum from http import HTTPStatus from types import MappingProxyType from typing import Final @@ -68,48 +69,104 @@ class _ErrorPayload(LiteLLMBaseModel): return {"message": self.error} if isinstance(self.error, str) else self.error -def _is_plain_mapping(value: object) -> TypeIs[Mapping[object, object]]: # guard-ok: exact type check - return type(value) in (dict, MappingProxyType) +_DROPPED: Final = object() + + +class _BaseForm(Enum): + PRINTED = "printed" + DUMPED = "dumped" + + +def _is_plain_dict(value: object) -> TypeIs[Mapping[object, object]]: # guard-ok: exact type check + return type(value) is dict def _is_plain_sequence(value: object) -> TypeIs[Sequence[object]]: # guard-ok: exact type check return type(value) in (list, tuple) -def _json_key(key: object) -> str | int | float | bool | None: - return key if key is None or isinstance(key, str | int | float | bool) else repr(key) +def _is_proxy(value: object) -> TypeIs[Mapping[object, object]]: # guard-ok: mappingproxy cannot be subclassed + return isinstance(value, MappingProxyType) -def _json_safe_items(items: Iterable[tuple[object, object]]) -> Mapping[str | int | float | bool | None, object]: - return {_json_key(key): _json_safe(item) for key, item in items} +def _prints_as_dict(proxy: Mapping[object, object]) -> bool: + return repr(proxy) == f"mappingproxy({dict(proxy)!r})" -def _json_safe(value: object) -> object: - if value is None or isinstance(value, str | int | float | bool): - return value - if _is_plain_mapping(value): - return _json_safe_items(value.items()) - if _is_plain_sequence(value): - return [_json_safe(item) for item in value] - if isinstance(value, BaseModel) and type(value).__repr__ is BaseModel.__repr__: - shown: Final[Iterable[tuple[str | None, object]]] = value.__repr_args__() - return _json_safe_items(shown) - return repr(value) +def _prints_its_fields(model: BaseModel) -> bool: + kind: Final = type(model) + return ( + kind.__repr__ is BaseModel.__repr__ + and kind.__str__ is BaseModel.__str__ + and kind.__repr_str__ is BaseModel.__repr_str__ + ) -def _stream_json(value: object) -> str | None: +def _str_skips_its_repr(model: BaseModel) -> bool: + kind: Final = type(model) + return kind.__str__ is BaseModel.__str__ and kind.__repr__ is not BaseModel.__repr__ + + +def _text(value: object, form: _BaseForm) -> object: try: - return json.dumps(_json_safe(value)) - except Exception: # noqa: BLE001 # hook values run their own __repr__, so any failure must fall back - return None + if form is _BaseForm.PRINTED: + return repr(value) + if isinstance(value, int): + return int(value) + if isinstance(value, float): + return float(value) + if isinstance(value, BaseModel) and _str_skips_its_repr(value): + return repr(value) + return str(value) + except Exception: # noqa: BLE001 # hook values run their own __repr__ and __str__, so a failure drops the value + return _DROPPED -def _encodable_items(error: Mapping[str, object]) -> Mapping[str, object]: - return {key: value for key, value in error.items() if _stream_json(value) is not None} +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 _expansion(value: object, form: _BaseForm) -> tuple[Iterable[tuple[object, object]], _BaseForm] | None: + if _is_plain_dict(value): + return value.items(), form + if _is_proxy(value) and _prints_as_dict(value): + return value.items(), _BaseForm.PRINTED + if isinstance(value, BaseModel) and _prints_its_fields(value): + return value.__repr_args__(), _BaseForm.PRINTED + return None + + +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 + if id(value) in path: + return _text(value, form) + inner_path: Final = path | frozenset((id(value),)) + try: + if _is_plain_sequence(value): + shown_items: Final = [_json_safe(item, form, inner_path) for item in value] + 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 responses_chunk_json(chunk: object) -> str | None: - return _stream_json(chunk) if _is_plain_mapping(chunk) else None + if not _is_plain_dict(chunk) and not _is_proxy(chunk): + return None + shown: Final = _json_safe(chunk, _BaseForm.PRINTED, frozenset()) + return None if isinstance(shown, str) or shown is _DROPPED else json.dumps(shown) + + +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: @@ -242,9 +299,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 = _stream_json(_error_event(fields, _encodable_items(error), sequence_number)) or json.dumps( - _error_event(fields, fields, sequence_number) - ) + payload: Final = json.dumps(_error_event(fields, _dumped_detail(error), sequence_number)) return f"event: error\ndata: {payload}\n\n" def format_error_message(self, frame: str) -> str: diff --git a/tests/unit/proxy/proxy_server/test_streaming_helpers.py b/tests/unit/proxy/proxy_server/test_streaming_helpers.py index 4dfb3daec52..e055d98761a 100644 --- a/tests/unit/proxy/proxy_server/test_streaming_helpers.py +++ b/tests/unit/proxy/proxy_server/test_streaming_helpers.py @@ -16,8 +16,12 @@ Pins covered: from __future__ import annotations import asyncio +import functools import json -from collections.abc import AsyncIterator, Callable, Mapping +from collections.abc import AsyncIterator, Callable, Iterable, 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 @@ -248,24 +252,118 @@ 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): + def __repr__(self) -> str: + return "_MaskedReprString([secure])" + + _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"), + _MaskedReprModel(token="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_model", + "masked_repr_str_model", + "masked_repr_args_model", +) + @pytest.mark.parametrize( - ("nested", "expected"), - [ - (_UPSTREAM_HEADERS, repr(_UPSTREAM_HEADERS)), - (_REDACTED_HEADERS, repr(_REDACTED_HEADERS)), - (MappingProxyType({"headers": _REDACTED_HEADERS}), {"headers": repr(_REDACTED_HEADERS)}), - (_HiddenFieldModel(user="rig", password="hook-canary"), {"user": "rig"}), - (_MaskedReprModel(token="hook-canary"), repr(_MaskedReprModel(token="hook-canary"))), - ], - ids=["header_mapping", "redacted_dict", "proxy_over_redacted_dict", "repr_false_field", "masked_repr_model"], + "nested", + [*_MASKED_IN_EVERY_FRAME, _MaskedReprString("hook-canary")], + ids=[*_MASKED_IN_EVERY_FRAME_IDS, "masked_repr_string"], ) -def test_responses_stream_frames_never_show_what_a_nested_value_hides_from_its_repr( - nested: object, expected: object +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 + + +@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 _RigColor(Enum): + RED = 1 + + +class _Score(float): + pass + + +_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)), + (HTTPStatus.BAD_REQUEST, repr(HTTPStatus.BAD_REQUEST), 400), + (_Score(0.5), repr(_Score(0.5)), 0.5), + ], + ids=["datetime", "enum", "int_enum", "float_subclass"], +) +def test_responses_stream_frames_print_values_json_cannot_take_as_base_printed_them( + nested: object, printed: str, dumped: object ) -> None: - assert _responses_frame_payload(nested)["nested"] == expected + 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 @pytest.mark.parametrize("mapping", [dict, MappingProxyType], ids=["dict", "mapping_proxy"]) @@ -290,39 +388,65 @@ def _self_referencing_chunk() -> Mapping[str, object]: return chunk +def test_responses_stream_frames_print_a_value_that_holds_itself_as_its_text() -> None: + nested: Final = _self_referencing_chunk() + + assert _responses_frame_payload(nested)["nested"] == {"type": "response.output_text.delta", "self": repr(nested)} + + 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( "chunk", [ - _self_referencing_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=["self_reference", "header_mapping_chunk", "redacted_dict_chunk"], + ids=["header_mapping_chunk", "redacted_dict_chunk", "proxy_wrapping_redacted_dict_chunk"], ) -def test_responses_sse_chunk_json_cannot_encode_streams_as_a_plain_data_frame(chunk: Mapping[str, object]) -> None: +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.parametrize( - "nested", - [_self_referencing_chunk(), _UnprintableValue()], - ids=["self_reference", "unprintable_value"], -) -def test_responses_error_event_drops_only_the_error_detail_json_cannot_encode(nested: object) -> None: - frame: Final = ResponsesStreamErrorState().format_error( - {"message": "blocked", "code": "400", "type": "guardrail_error", "nested": nested} - ) - - assert frame.startswith("event: error\ndata: ") - payload: Final = json.loads(frame.split("data: ", 1)[1]) - assert (payload["type"], payload["message"]) == ("error", "blocked") - assert payload["error"] == {"message": "blocked", "code": "400", "type": "guardrail_error"} - @pytest.mark.asyncio @pytest.mark.parametrize( ("guardrail_error", "error_object"),