fix(proxy): never fail a Responses stream frame on a value JSON cannot encode

Only dicts and read-only dict views expand into JSON, at the top level too, so a header mapping yielded as a whole chunk keeps its masked string form. Any failure while encoding a chunk falls back to the plain data frame the base sent, and an error event that cannot encode still goes out with its code and message, so a late block never drops the connection.

litellm/proxy/common_utils/responses_stream_errors.py

@@ -68,10 +68,10 @@ class _ErrorPayload(LiteLLMBaseModel):
         return {"message": self.error} if isinstance(self.error, str) else self.error
 
 
-def _is_mapping(
+def _is_dict_like(
     value: object,
-) -> TypeIs[Mapping[object, object]]:  # guard-ok: isinstance decides it, keys and values stay object
-    return isinstance(value, Mapping)
+) -> TypeIs[dict[object, object] | MappingProxyType[object, object]]:  # guard-ok: isinstance decides it
+    return isinstance(value, dict | MappingProxyType)
 
 
 def _is_mapping_proxy(
@@ -96,15 +96,17 @@ def _jsonable_stream_value(value: object) -> object:
     return str(value)
 
 
-def responses_chunk_json(chunk: object) -> str | None:
-    if not _is_mapping(chunk):
-        return None
+def _stream_json(value: Mapping[object, object]) -> str | None:
     try:
-        return json.dumps(_with_json_keys(chunk), default=_jsonable_stream_value)
-    except (TypeError, ValueError):
+        return json.dumps(_with_json_keys(value), default=_jsonable_stream_value)
+    except Exception:  # noqa: BLE001  # hook values run their own __str__ and serializers, so any failure must fall back
         return None
 
 
+def responses_chunk_json(chunk: object) -> str | None:
+    return _stream_json(chunk) if _is_dict_like(chunk) else None
+
+
 def sse_error_object(data: str | bytes) -> Mapping[str, object] | None:
     try:
         return _ErrorPayload.model_validate_json(data).error_object()
@@ -221,15 +223,24 @@ class ResponsesStreamErrorState:
         return prefix.encode("utf-8") + frame if isinstance(frame, bytes) else prefix + frame
 
     def format_error(self, error: Mapping[str, object]) -> str:
-        payload: Final = json.dumps(
+        fields: Final = responses_error_fields(error)
+        sequence_number: Final = self.sequence_number + 1
+        payload: Final = _stream_json(
             {
                 "type": ResponsesAPIStreamEvents.ERROR.value,
-                **responses_error_fields(error),
+                **fields,
                 "param": error.get("param"),
-                "sequence_number": self.sequence_number + 1,
+                "sequence_number": sequence_number,
                 "error": error,
-            },
-            default=_jsonable_stream_value,
+            }
+        ) or json.dumps(
+            {
+                "type": ResponsesAPIStreamEvents.ERROR.value,
+                **fields,
+                "param": None,
+                "sequence_number": sequence_number,
+                "error": fields,
+            }
         )
         return f"event: error\ndata: {payload}\n\n"
 

tests/unit/proxy/proxy_server/test_streaming_helpers.py

@@ -258,14 +258,42 @@ def _self_referencing_chunk() -> Mapping[str, object]:
     return chunk
 
 
+class _UnprintableValue:
+    def __str__(self) -> str:
+        raise RuntimeError("no string form")
+
+
 @pytest.mark.parametrize(
     "chunk",
-    [{"type": "response.output_text.delta", "meta": {("a", "b"): 1}}, _self_referencing_chunk()],
-    ids=["nested_dict_tuple_key", "self_reference"],
+    [
+        {"type": "response.output_text.delta", "meta": {("a", "b"): 1}},
+        _self_referencing_chunk(),
+        {"type": "response.output_text.delta", "meta": _UnprintableValue()},
+        _UPSTREAM_HEADERS,
+    ],
+    ids=["nested_dict_tuple_key", "self_reference", "unprintable_value", "header_mapping_chunk"],
 )
 def test_responses_sse_chunk_json_cannot_encode_streams_as_a_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",
+    [
+        MappingProxyType({"bad": {("a", "b"): 1}}),
+        _self_referencing_chunk(),
+        _UnprintableValue(),
+    ],
+    ids=["proxy_over_tuple_key", "self_reference", "unprintable_value"],
+)
+def test_responses_error_event_json_cannot_encode_still_ends_the_stream_with_its_message(nested: object) -> None:
+    frame: Final = ResponsesStreamErrorState().format_error({"message": "blocked", "code": 400, "nested": nested})
+
+    assert frame.startswith("event: error\ndata: ")
+    payload: Final = json.loads(frame.split("data: ", 1)[1])
+    assert (payload["type"], payload["message"], payload["error"]["message"]) == ("error", "blocked", "blocked")
+    assert payload["error"]["code"] == payload["code"]
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     ("guardrail_error", "error_object"),