Merge remote-tracking branch 'origin/litellm_internal_staging' into litellm_fix_branchless_provider_status_mapping

# Conflicts:
#	tests/test_litellm/litellm_core_utils/test_exception_mapping_utils.py

litellm/litellm_core_utils/exception_mapping_utils.py

@@ -550,6 +550,13 @@ def _map_anthropic_exception(
                 llm_provider="anthropic",
                 model=model,
             )
+        elif original_exception.status_code == 403:
+            raise PermissionDeniedError(
+                message=f"AnthropicException - {error_str}",
+                llm_provider="anthropic",
+                model=model,
+                response=original_exception.response,
+            )
         elif original_exception.status_code == 400 or original_exception.status_code == 413:
             raise BadRequestError(
                 message=f"AnthropicException - {error_str}",

litellm/llms/anthropic/experimental_pass_through/messages/handler.py

@@ -12,6 +12,7 @@ from functools import partial
 from typing import Any, Final, cast
 
 import litellm
+from litellm.litellm_core_utils.exception_mapping_utils import exception_type
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.llms.anthropic.common_utils import (
     flatten_unencrypted_web_search_results_in_anthropic_messages,
@@ -21,6 +22,7 @@ from litellm.llms.anthropic.common_utils import (
 from litellm.llms.base_llm.anthropic_messages.transformation import (
     BaseAnthropicMessagesConfig,
 )
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
 from litellm.types.llms.anthropic_messages.anthropic_request import AnthropicMetadata
@@ -382,13 +384,18 @@ async def anthropic_messages(
     )
     ctx: Final = contextvars.copy_context()
     func_with_context: Final = partial(ctx.run, func)
-    init_response: Final = await loop.run_in_executor(None, func_with_context)
-
-    if asyncio.iscoroutine(init_response):
-        response = await init_response
-    else:
-        response = init_response
-    return response
+    try:
+        init_response: Final = await loop.run_in_executor(None, func_with_context)
+        if asyncio.iscoroutine(init_response):
+            return await init_response
+        return init_response
+    except BaseLLMException as e:
+        raise exception_type(
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            original_exception=e,
+            extra_kwargs=kwargs,
+        )
 
 
 def validate_anthropic_api_metadata(metadata: dict | None = None) -> dict | None:

tests/e2e/llm_translation/test_together_ai_e2e.py

@@ -5,14 +5,19 @@ in the proxy's own cost map that carries both capability flags. Two backends are
 pinned because the registry has no flag for what they prove: ``enable_thinking`` is a
 Qwen chat-template contract, and MiniMax-M3 is the serverless model whose template
 renders a replayed ``reasoning_content`` back into the prompt (Qwen and DeepSeek
-silently drop it). Requires TOGETHER_API_KEY on the proxy; no skip gate.
+silently drop it). MiniMax-M3 honors that replayed field on nearly every call, not
+every call (one miss in dozens of otherwise identical calls), so the replay case asks
+up to ``REPLAY_ATTEMPTS`` times and fails only when no answer carries the secret, which
+a proxy that strips the field guarantees. Requires TOGETHER_API_KEY on the proxy; no
+skip gate.
 """
 
 from __future__ import annotations
 
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
 from dataclasses import dataclass
 from datetime import date
+from typing import Final
 
 import pytest
 from e2e_config import unique_marker
@@ -51,6 +56,7 @@ REASONING_REPLAY_BACKEND = "together_ai/MiniMaxAI/MiniMax-M3"
 SECRET_PROMPT = "Remember this for later and reply with just OK."
 SECRET_REASONING = "The user told me their favorite color is chartreuse. I must remember it."
 SECRET_QUESTION = "What is my favorite color? Answer with one word."
+REPLAY_ATTEMPTS: Final = 3
 
 ARITHMETIC_PROMPT = "What is 17 + 26? Answer with just the number."
 WEATHER_PROMPT = "What is the weather in Paris? Use the tool."
@@ -179,6 +185,18 @@ def _message(response: ChatResponse) -> OutMessage:
     return message
 
 
+def _carries_secret(answer: OutMessage) -> bool:
+    return answer.content is not None and "chartreuse" in answer.content.lower()
+
+
+def _answers_until_secret(client: PassthroughClient, key: str, body: ChatBody) -> Iterator[OutMessage]:
+    answers: Final = (_message(unwrap(client.proxy.chat(key, body))) for _ in range(REPLAY_ATTEMPTS))
+    for answer in answers:
+        yield answer
+        if _carries_secret(answer):
+            return
+
+
 def _deltas(result: StreamingResponse) -> list[_StreamDelta]:
     require_successful_call(result)
     assert result.is_streaming, f"response was not streamed: {result.headers}"
@@ -376,25 +394,19 @@ class TestTogetherChatCompletions:
         self, client: PassthroughClient, resources: ResourceManager
     ) -> None:
         model, key = _register(client, resources, REASONING_REPLAY_BACKEND)
-
-        answer = _message(
-            unwrap(
-                client.proxy.chat(
-                    key,
-                    ChatBody(
-                        model=model,
-                        messages=[
-                            ChatMessage(role="user", content=SECRET_PROMPT),
-                            ChatAssistantTurn(content="OK.", reasoning_content=SECRET_REASONING),
-                            ChatMessage(role="user", content=SECRET_QUESTION),
-                        ],
-                        max_tokens=512,
-                    ),
-                )
-            )
+        body: Final = ChatBody(
+            model=model,
+            messages=[
+                ChatMessage(role="user", content=SECRET_PROMPT),
+                ChatAssistantTurn(content="OK.", reasoning_content=SECRET_REASONING),
+                ChatMessage(role="user", content=SECRET_QUESTION),
+            ],
+            max_tokens=512,
         )
-        assert answer.content and "chartreuse" in answer.content.lower(), (
-            f"the replayed reasoning_content never reached Together: {answer}"
+
+        answers: Final = tuple(_answers_until_secret(client, key, body))
+        assert any(_carries_secret(answer) for answer in answers), (
+            f"the replayed reasoning_content never reached Together in {len(answers)} attempts: {answers}"
         )
 
     @pytest.mark.covers("llm.chat_completions.together_ai.basic.nonstream.cost_logged")

tests/e2e/logging/test_otel_trace_e2e.py

@@ -743,3 +743,61 @@ class TestOtelTraceCompleteness:
         )
         genai = next(span for span in hits[0].spans if span.operation_name == genai_span)
         _assert_error_span_contract(genai)
+
+    @pytest.mark.covers("logging.otel.failure.exports_metric", exercised_on=["messages"])
+    def test_failed_messages_error_span_attributes(
+        self, client: LoggingClient, otel_reader: OtelReader, resources: ResourceManager
+    ) -> None:
+        """A failed `/v1/messages` request must carry the same error-span
+        contract as a failed `/chat/completions` request (LIT-6164). The
+        async messages entrypoint used to surface the provider handler's raw
+        BaseLLMException to the failure logger, so the model-call span came
+        out with error.type=BaseLLMException and no
+        litellm.provider.error.llm_provider attribute.
+
+        Same setup as the chat sibling: a deployment with an invalid upstream
+        API key passes proxy auth and fails at the provider with a real 401,
+        and failed requests are not billed, so no cost-write span."""
+        route = "/v1/messages"
+        _assert_otel_destination_configured(client)
+
+        model_name = f"otel-err-{unique_marker()}"
+        model_id = client.create_model(
+            model_name,
+            LiteLLMParamsBody(model="anthropic/claude-haiku-4-5", api_key=INVALID_UPSTREAM_API_KEY),
+        )
+        resources.defer(lambda: client.delete_model(model_id))
+        key = client.key_with_alias(f"otel-err-{unique_marker()}", models=[model_name])
+        resources.defer(lambda: client.delete_key(key))
+
+        deadline = time.monotonic() + client.proxy.poll_timeout
+        while True:
+            outcome = client.messages_raw(key, model_name, "trigger an upstream auth failure", max_tokens=16)
+            assert not outcome.ok, "the call must fail; the deployment's upstream key is invalid"
+            if "AnthropicException" in outcome.body or time.monotonic() >= deadline:
+                break
+            time.sleep(client.proxy.poll_interval)
+        assert "AnthropicException" in outcome.body, (
+            "never saw the mapped upstream provider failure before the deadline; either the key is "
+            "still propagating or the messages route surfaced the raw unmapped provider error - "
+            f"last outcome {outcome.status_code}: {outcome.body[:200]}"
+        )
+        assert outcome.status_code == 401, (
+            f"an upstream auth failure must map to 401, got {outcome.status_code}: {outcome.body[:200]}"
+        )
+        assert outcome.call_id is not None, "failed responses must still carry x-litellm-call-id"
+
+        genai_span = f"chat {model_name}"
+        hits = otel_reader.poll_traces_for_call(
+            call_id=outcome.call_id,
+            settled_names=_settled_names(route=route, genai_span=genai_span, require_cost_span=False),
+            settled_prefixes={DB_SPAN_PREFIX},
+        )
+        _assert_complete_trace(hits, route=route, genai_span=genai_span, require_cost_span=False)
+
+        root = next(span for span in hits[0].spans if not span.references)
+        assert str(_tag(root, "http.status_code")) == "401", (
+            f"the SERVER span must record the 401 the client received, got {_tag(root, 'http.status_code')!r}"
+        )
+        genai = next(span for span in hits[0].spans if span.operation_name == genai_span)
+        _assert_error_span_contract(genai)

tests/test_litellm/llms/anthropic/experimental_pass_through/messages/test_anthropic_experimental_pass_through_messages_handler.py

@@ -7,6 +7,7 @@ from typing import Any, Dict, List
 import httpx
 import pytest
 from fastapi.testclient import TestClient
+from pydantic import ValidationError
 
 
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -1286,3 +1287,96 @@ class TestMessagesStreamingSuccessLogging:
         assert payload["call_type"] == "acompletion"
         assert payload["total_tokens"] > 0
         assert payload["response_cost"] > 0
+
+
+class _FailureCapture(CustomLogger):
+    def __init__(self):
+        super().__init__()
+        self.error_information: List[Dict[str, Any]] = []
+
+    async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
+        payload = kwargs.get("standard_logging_object") or {}
+        self.error_information.append(payload.get("error_information") or {})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "upstream_status, upstream_error_type, expected_exception",
+    [
+        (401, "authentication_error", litellm.AuthenticationError),
+        (403, "permission_error", litellm.PermissionDeniedError),
+    ],
+)
+async def test_anthropic_messages_maps_provider_exception_before_failure_logging(
+    monkeypatch, upstream_status, upstream_error_type, expected_exception
+):
+    """Regression test for LIT-6164. The async /v1/messages entrypoint awaited the
+    provider handler without exception_type mapping, so the @client failure
+    handler (and every logger behind it, e.g. OTel error spans) saw the raw
+    BaseLLMException: error.type=BaseLLMException and no llm_provider.
+
+    The 403 row pins the upstream status on the way through the mapper: Anthropic's
+    documented permission_error must reach the caller as a 403, never as the mapper's
+    APIConnectionError 500 fallthrough."""
+    from litellm.llms.anthropic.experimental_pass_through.messages import handler
+
+    capture = _FailureCapture()
+    monkeypatch.setattr(litellm, "callbacks", [capture])
+
+    def upstream_rejects_the_request(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(
+            upstream_status,
+            json={"type": "error", "error": {"type": upstream_error_type, "message": "rejected upstream"}},
+            request=request,
+        )
+
+    upstream = AsyncHTTPHandler()
+    upstream.client = httpx.AsyncClient(transport=httpx.MockTransport(upstream_rejects_the_request))
+
+    with pytest.raises(expected_exception) as excinfo:
+        await handler.anthropic_messages(
+            max_tokens=16,
+            messages=[{"role": "user", "content": "hi"}],
+            model="anthropic/claude-haiku-4-5",
+            custom_llm_provider="anthropic",
+            api_key="sk-invalid",
+            client=upstream,
+        )
+
+    assert excinfo.value.status_code == upstream_status
+    assert excinfo.value.llm_provider == "anthropic"
+    assert "AnthropicException" in excinfo.value.message
+    assert f'"{upstream_error_type}"' in excinfo.value.message
+
+    assert capture.error_information, "the failure handler must have logged the mapped exception"
+    error_information = capture.error_information[0]
+    assert error_information.get("error_class") == expected_exception.__name__
+    assert error_information.get("llm_provider") == "anthropic"
+    assert error_information.get("error_code") == str(upstream_status)
+
+
+@pytest.mark.asyncio
+async def test_anthropic_messages_leaves_non_provider_failures_unmapped():
+    """The mapping boundary is for provider failures only. A request rejected before
+    the provider call (here invalid metadata) must surface as the original exception,
+    not as the mapper's APIConnectionError, whose message embeds a server traceback."""
+    from litellm.llms.anthropic.experimental_pass_through.messages import handler
+
+    def upstream_must_not_be_called(request: httpx.Request) -> httpx.Response:
+        raise AssertionError("the provider must not be called for a request rejected locally")
+
+    upstream = AsyncHTTPHandler()
+    upstream.client = httpx.AsyncClient(transport=httpx.MockTransport(upstream_must_not_be_called))
+
+    with pytest.raises(ValidationError) as excinfo:
+        await handler.anthropic_messages(
+            max_tokens=16,
+            messages=[{"role": "user", "content": "hi"}],
+            model="anthropic/claude-haiku-4-5",
+            custom_llm_provider="anthropic",
+            api_key="sk-invalid",
+            client=upstream,
+            metadata={"user_id": 123},
+        )
+
+    assert "Traceback" not in str(excinfo.value)