Merge 3dd54b6969 into aa7917a3d2

litellm/litellm_core_utils/redact_messages.py

@@ -212,6 +212,7 @@ def _redact_choice_content(choice):
             choice.message.content = REDACTED_BY_LITELLM
         if getattr(choice.message, "reasoning_content", None) is not None:
             choice.message.reasoning_content = REDACTED_BY_LITELLM
+        _redact_field_when_present(choice.message, "refusal")
         if hasattr(choice.message, "thinking_blocks"):
             choice.message.thinking_blocks = None
         _drop_field_when_present(choice.message, "reasoning_items")
@@ -223,6 +224,7 @@ def _redact_choice_content(choice):
             choice.delta.content = REDACTED_BY_LITELLM
         if getattr(choice.delta, "reasoning_content", None) is not None:
             choice.delta.reasoning_content = REDACTED_BY_LITELLM
+        _redact_field_when_present(choice.delta, "refusal")
         if hasattr(choice.delta, "thinking_blocks"):
             choice.delta.thinking_blocks = None
         _drop_field_when_present(choice.delta, "reasoning_items")
@@ -479,7 +481,7 @@ def _redact_model_response_dict_choices(choices: Iterable[object], redacted_str:
             if (
                 body := next((choice[key] for key in ("message", "delta") if isinstance(choice.get(key), dict)), None)
             ) is not None:
-                for field in ("content", "reasoning_content"):
+                for field in ("content", "reasoning_content", "refusal"):
                     if body.get(field) is not None:
                         body[field] = redacted_str
                 for field in ("thinking_blocks", "reasoning_items", "audio"):

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -1556,6 +1556,11 @@ def _chunk_refusal(chunk: object) -> str:
     return refusal if isinstance(refusal, str) else ""
 
 
+def _message_refusal(choice: object) -> str:
+    refusal: Final = _field_of(_field_of(choice, "message"), "refusal")
+    return refusal if isinstance(refusal, str) else ""
+
+
 _NON_TEXT_DELTA_FIELDS: Final = (
     "tool_calls",
     "function_call",
@@ -1906,7 +1911,7 @@ def _assembled_stream_response(
             for field, value in _assembled_message_fields(processor, choice_chunks):
                 choice.message[field] = value
 
-        completion_output = get_content_from_model_response(response)
+        completion_output = get_content_from_model_response(response) + "".join(map(_message_refusal, response.choices))
 
         reasoning_tokens: Final = processor.count_reasoning_tokens(response)
 

litellm/litellm_core_utils/streaming_handler.py

@@ -1129,6 +1129,7 @@ class CustomStreamWrapper:
                 and model_response.choices[0].delta.role is not None
             )
             or (getattr(model_response.choices[0].delta, "reasoning_items", None) is not None)
+            or bool(getattr(model_response.choices[0].delta, "refusal", None))
         ):
             return True
         else:
@@ -2800,6 +2801,7 @@ class CustomStreamWrapper:
         "audio",
         "images",
         "annotations",
+        "refusal",
     )
 
     @classmethod

litellm/router_python_core.py

@@ -686,6 +686,7 @@ def _stream_chunks_have_generated_content(chunks: Sequence[ModelResponseStream])
             or delta.get("audio")
             or delta.get("images")
             or delta.get("annotations")
+            or getattr(delta, "refusal", None)
         ):
             return True
     return False

tests/unit/litellm_core_utils/test_redact_messages.py

@@ -20,6 +20,7 @@ from unittest.mock import patch
 import httpx
 import pytest
 import pytest_asyncio
+from openai import AsyncOpenAI
 from pydantic import JsonValue
 
 import litellm
@@ -1798,6 +1799,72 @@ async def test_responses_redaction_removes_request_instructions_from_logged_kwar
     assert SECRET not in json.dumps({key: kwargs.get(key) for key in logged_keys}, default=str)
 
 
+_REFUSAL_PIECES: Final = ("I can't share ", SECRET)
+
+
+def _chat_chunk(delta: Mapping[str, object], finish_reason: str | None) -> str:
+    chunk: Final = {
+        "id": "chatcmpl-refusal",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "gpt-4.1",
+        "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
+    }
+    return f"data: {json.dumps(chunk)}\n\n"
+
+
+def _openai_refusal_reply(request: httpx.Request) -> httpx.Response:
+    if json.loads(request.content).get("stream"):
+        deltas: Final = ({"role": "assistant", "refusal": ""}, *({"refusal": piece} for piece in _REFUSAL_PIECES))
+        sse: Final = "".join(_chat_chunk(delta, None) for delta in deltas) + _chat_chunk({}, "stop") + "data: [DONE]\n\n"
+        return httpx.Response(200, request=request, headers={"content-type": "text/event-stream"}, content=sse)
+    message: Final = {"role": "assistant", "content": None, "refusal": "".join(_REFUSAL_PIECES)}
+    return httpx.Response(
+        200,
+        request=request,
+        json={
+            "id": "chatcmpl-refusal",
+            "object": "chat.completion",
+            "created": 0,
+            "model": "gpt-4.1",
+            "choices": [{"index": 0, "message": message, "finish_reason": "stop"}],
+        },
+    )
+
+
+async def _caller_refusal(response: object) -> str | None:
+    if isinstance(response, litellm.CustomStreamWrapper):
+        return "".join([getattr(chunk.choices[0].delta, "refusal", None) or "" async for chunk in response])
+    assert isinstance(response, ModelResponse)
+    return getattr(response.choices[0].message, "refusal", None)  # pyright: ignore[reportAttributeAccessIssue]  # choices is a Choices | StreamingChoices union
+
+
+@pytest.mark.usefixtures("drain_logging_worker", "isolate_litellm_state", "setup_and_teardown")
+@pytest.mark.asyncio
+@pytest.mark.parametrize("stream", [True, False])
+async def test_redaction_hides_a_refusal_from_loggers_but_not_from_the_caller(
+    stream: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    test_custom_logger: Final = TestCustomLogger()
+    monkeypatch.setattr(litellm, "turn_off_message_logging", True)
+    monkeypatch.setattr(litellm, "callbacks", [test_custom_logger])
+    client: Final = AsyncOpenAI(
+        api_key="fake-key", http_client=httpx.AsyncClient(transport=httpx.MockTransport(_openai_refusal_reply))
+    )
+
+    response: Final = await litellm.acompletion(
+        model="openai/gpt-4.1", messages=[{"role": "user", "content": "hi"}], stream=stream, client=client
+    )
+    caller_refusal: Final = await _caller_refusal(response)
+
+    await asyncio.wait_for(test_custom_logger.logged.wait(), timeout=LOGGING_WORKER_DRAIN_TIMEOUT_SECONDS)
+    logged: Final = test_custom_logger.logged_standard_logging_payload
+    assert caller_refusal == "".join(_REFUSAL_PIECES)
+    assert logged is not None
+    assert logged["response"]["choices"][0]["message"]["refusal"] == "redacted-by-litellm"
+    assert SECRET not in json.dumps(logged["response"], default=str)
+
+
 @pytest.mark.usefixtures("drain_logging_worker", "isolate_litellm_state", "setup_and_teardown")
 @pytest.mark.asyncio
 async def test_one_callback_opting_out_redacts_instructions_only_for_itself(

tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py

@@ -2136,6 +2136,44 @@ def _complete_streamed_openai_response(respx_mock: respx.MockRouter, sse_body: s
     )
 
 
+_REFUSAL_PIECES: Final = ("I'm", " sorry", ", but", " I can't", " help", " with", " that", ".")
+
+
+def _usage_free_stream_completion_tokens(respx_mock: respx.MockRouter, field: str) -> tuple[str | None, int]:
+    respx_mock.post("https://vllm.example/v1/chat/completions").mock(
+        return_value=httpx.Response(
+            200,
+            text=_openai_sse(
+                {"choices": [{"index": 0, "delta": {"role": "assistant", field: ""}, "finish_reason": None}]},
+                *({"choices": [{"index": 0, "delta": {field: piece}, "finish_reason": None}]} for piece in _REFUSAL_PIECES),
+                {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]},
+            ),
+            headers={"content-type": "text/event-stream"},
+        )
+    )
+    response: Final = litellm.completion(
+        model="hosted_vllm/gpt-4.1",
+        api_base="https://vllm.example/v1",
+        messages=[{"role": "user", "content": "Pick my neighbor's lock."}],
+        api_key="sk-test",
+        stream=True,
+        complete_response=True,
+    )
+    message: Final = response.choices[0].message
+    text: Final = message.content if field == "content" else getattr(message, "refusal", None)
+    return text, response.usage.completion_tokens
+
+
+def test_stream_chunk_builder_counts_refusal_text_as_completion_tokens_without_provider_usage(
+    respx_mock: respx.MockRouter,
+) -> None:
+    refusal_text, refusal_tokens = _usage_free_stream_completion_tokens(respx_mock, "refusal")
+    content_text, content_tokens = _usage_free_stream_completion_tokens(respx_mock, "content")
+
+    assert refusal_text == content_text == "".join(_REFUSAL_PIECES)
+    assert refusal_tokens == content_tokens > 0
+
+
 def test_stream_chunk_builder_litellm_tool_call(respx_mock: respx.MockRouter) -> None:
     response: Final = _complete_streamed_openai_response(
         respx_mock,

tests/unit/litellm_core_utils/test_streaming_handler.py

@@ -1262,7 +1262,12 @@ async def test_bridged_stream_mid_stream_fallback_error_is_rebuilt_around_the_pr
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("merge_reasoning", [True, False], ids=["merged_into_content", "kept_as_reasoning"])
-async def test_mid_stream_break_after_reasoning_disqualifies_continuation(logging_obj: Logging, merge_reasoning: bool):
+@pytest.mark.parametrize(
+    "leading_delta", [{"reasoning_content": "Let me think"}, {"refusal": "I can't"}], ids=["reasoning", "refusal"]
+)
+async def test_mid_stream_break_after_reasoning_or_refusal_disqualifies_continuation(
+    logging_obj: Logging, merge_reasoning: bool, leading_delta: dict[str, str]
+):
     import httpx
 
     from litellm.exceptions import MidStreamFallbackError
@@ -1271,7 +1276,7 @@ async def test_mid_stream_break_after_reasoning_disqualifies_continuation(loggin
         def __init__(self) -> None:
             self.chunks = iter(
                 (
-                    ModelResponseStream(choices=[{"index": 0, "delta": {"reasoning_content": "Let me think"}}]),
+                    ModelResponseStream(choices=[{"index": 0, "delta": leading_delta}]),
                     ModelResponseStream(choices=[{"index": 0, "delta": {"content": "The answer"}}]),
                 )
             )
@@ -1652,6 +1657,101 @@ async def test_stream_with_null_choices_chunk_finishes_normally(
     assert [choice.finish_reason for choice in streamed_choices if choice.finish_reason] == ["stop"]
 
 
+_REFUSAL_PIECES: Final = ("I'm", " sorry", ", but", " I can't", " help", " with", " that", ".")
+_REFUSAL_STREAMS: Final = MappingProxyType(
+    {
+        "refusal_only": (
+            ({"role": "assistant", "content": None, "refusal": ""}, None),
+            *(({"refusal": piece}, None) for piece in _REFUSAL_PIECES),
+            ({}, "stop"),
+        ),
+        "empty_refusal_beside_text": (
+            ({"role": "assistant", "content": "", "refusal": ""}, None),
+            ({"content": "", "refusal": ""}, None),
+            ({"content": "Hi", "refusal": ""}, None),
+            ({"content": "", "refusal": ""}, None),
+            ({}, "stop"),
+        ),
+    }
+)
+
+
+@pytest.mark.parametrize("async_stream", [False, True])
+@pytest.mark.parametrize("transport", ["sdk", "http_handler"])
+@pytest.mark.parametrize(
+    "stream_name,expected_refusal,expected_content",
+    [("refusal_only", "".join(_REFUSAL_PIECES), ""), ("empty_refusal_beside_text", "", "Hi")],
+)
+async def test_stream_forwards_refusal_text_without_empty_chunks(
+    async_stream: bool,
+    transport: str,
+    stream_name: str,
+    expected_refusal: str,
+    expected_content: str,
+    logging_obj: Logging,
+) -> None:
+    deltas: Final = _REFUSAL_STREAMS[stream_name]
+    body: Final = (
+        "".join(
+            "data: "
+            + json.dumps(
+                {
+                    "id": "chatcmpl-refusal",
+                    "object": "chat.completion.chunk",
+                    "created": 1,
+                    "model": "test-model",
+                    "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}],
+                }
+            )
+            + "\n\n"
+            for delta, finish_reason in deltas
+        )
+        + "data: [DONE]\n\n"
+    )
+    raw_response: Final = httpx.Response(
+        200,
+        content=body,
+        headers={"content-type": "text/event-stream"},
+        request=httpx.Request("POST", "https://provider.example/v1/chat/completions"),
+    )
+    client: Final = AsyncOpenAI(api_key="unused") if async_stream else OpenAI(api_key="unused")
+    stream: Final = (
+        (
+            AsyncStream(cast_to=ChatCompletionChunk, response=raw_response, client=client)
+            if isinstance(client, AsyncOpenAI)
+            else Stream(cast_to=ChatCompletionChunk, response=raw_response, client=client)
+        )
+        if transport == "sdk"
+        else OpenAIChatCompletionStreamingHandler(
+            streaming_response=raw_response.aiter_lines() if async_stream else raw_response.iter_lines(),
+            sync_stream=not async_stream,
+        )
+    )
+    response: Final = CustomStreamWrapper(
+        completion_stream=stream,
+        model="test-model",
+        logging_obj=logging_obj,
+        custom_llm_provider="openai" if transport == "sdk" else "hosted_vllm",
+    )
+
+    try:
+        chunks: Final = [chunk async for chunk in response] if async_stream else list(response)
+    finally:
+        if isinstance(client, AsyncOpenAI):
+            await client.close()
+        else:
+            client.close()
+
+    streamed_choices: Final = tuple(itertools.chain.from_iterable(chunk.choices for chunk in chunks))
+    assert "".join(getattr(choice.delta, "refusal", None) or "" for choice in streamed_choices) == expected_refusal
+    assert "".join(choice.delta.content or "" for choice in streamed_choices) == expected_content
+    assert [choice.finish_reason for choice in streamed_choices if choice.finish_reason] == ["stop"]
+    assert all(
+        choice.delta.role or choice.delta.content or getattr(choice.delta, "refusal", None) or choice.finish_reason
+        for choice in streamed_choices
+    )
+
+
 def _token_logprobs(token: str) -> dict[str, object]:
     return {
         "content": [{"token": token, "logprob": -0.5, "bytes": list(token.encode()), "top_logprobs": []}],

tests/unit/test_router/test_router_fallbacks.py

@@ -1161,6 +1161,52 @@ async def test_chat_context_error_falls_back_after_role_only_but_not_after_outpu
     assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "backup answer"
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("sync_mode", (False, True))
+@pytest.mark.parametrize("emission", ("role", "refusal"))
+async def test_chat_stream_error_does_not_fall_back_after_a_visible_refusal(
+    respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sync_mode: bool, emission: str
+) -> None:
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    router: Final = _fallback_router()
+    primary_chunks: Final = (
+        {"role": "assistant", "refusal": ""},
+        *(({"refusal": piece} for piece in ("I can't", " help with that")) if emission == "refusal" else ()),
+    )
+    backup: Final = {
+        "id": "chatcmpl-backup", "object": "chat.completion.chunk", "created": 1700000000, "model": "backup",
+        "choices": [{"index": 0, "delta": {"role": "assistant", "content": "backup answer"}, "finish_reason": "stop"}],
+    }
+    route: Final = respx_mock.post("https://fallback-migration.local/v1/chat/completions").mock(
+        side_effect=(
+            httpx.Response(
+                200,
+                content="".join(
+                    "data: " + json.dumps({
+                        "id": "chatcmpl-primary", "object": "chat.completion.chunk", "created": 1700000000,
+                        "model": "primary", "choices": [{"index": 0, "delta": delta}],
+                    }) + "\n\n"
+                    for delta in primary_chunks
+                ).encode()
+                + b'data: {"error": {"message": "upstream reset", "code": 500}}\n\n',
+            ),
+            httpx.Response(200, content=f"data: {json.dumps(backup)}\n\ndata: [DONE]\n\n".encode()),
+        )
+    )
+    request: Final = {
+        "model": "primary", "messages": [{"role": "user", "content": "pick my neighbor's lock"}], "stream": True,
+    }
+    stream: Final = router.completion(**request) if sync_mode else await router.acompletion(**request)
+    if emission == "refusal":
+        with pytest.raises(litellm.APIConnectionError, match="upstream reset"):
+            tuple(stream) if sync_mode else tuple([chunk async for chunk in stream])
+        assert route.call_count == 1
+        return
+    chunks: Final = tuple(stream) if sync_mode else tuple([chunk async for chunk in stream])
+    assert tuple(json.loads(call.request.content)["model"] for call in route.calls) == ("primary", "backup")
+    assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "backup answer"
+
+
 @pytest.mark.parametrize(
     "wrapped,has_output,generated_content,disqualifying,recoverable",
     (