Merge e9d6cde513 into e0982d6248

litellm/rust_bridge/failures.py

@@ -17,8 +17,8 @@ _UPSTREAM_HEADERS: Final = TypeAdapter(list[tuple[str, str]])
 
 class UpstreamFailure(Exception):
     def __init__(self, response: httpx.Response, cause: Exception) -> None:
-        super().__init__(str(cause))
-        self.message: Final = str(cause)
+        super().__init__(response.text)
+        self.message: Final = response.text
         self.response: Final = response
         self.status_code: Final = response.status_code
         self.__cause__ = cause
@@ -49,13 +49,20 @@ class ExceptionMapper(Protocol):
     ) -> Exception: ...
 
 
-def map_failure(error: Exception, model: str, request_provider: str, kwargs: Mapping[str, object]) -> Exception:
+def map_failure(
+    error: Exception,
+    model: str,
+    request_provider: str,
+    kwargs: Mapping[str, object],
+    *,
+    preserve_model_prefix: bool = False,
+) -> Exception:
     mapper: Final = cast(  # cast-ok: bounded adapter for the legacy public exception mapper
         ExceptionMapper, litellm.exception_type
     )
     try:
         return mapper(
-            model=model.removeprefix(f"{request_provider}/"),
+            model=model if preserve_model_prefix else model.removeprefix(f"{request_provider}/"),
             custom_llm_provider=request_provider,
             original_exception=error,
             completion_kwargs=dict(kwargs),
@@ -67,11 +74,19 @@ def map_failure(error: Exception, model: str, request_provider: str, kwargs: Map
 
 
 def map_native_failure(
-    error: Exception, model: str, request_provider: str, kwargs: Mapping[str, object], api_base: str | None = None
+    error: Exception,
+    model: str,
+    request_provider: str,
+    kwargs: Mapping[str, object],
+    api_base: str | None = None,
+    *,
+    preserve_model_prefix: bool = False,
 ) -> Exception:
     """`map_failure`, reading a native `(status, body)` provider failure as the HTTP response it was."""
     original: Final = _upstream_failure(error, api_base)
-    public_error: Final = map_failure(original, model, request_provider, kwargs)
+    public_error: Final = map_failure(
+        original, model, request_provider, kwargs, preserve_model_prefix=preserve_model_prefix
+    )
     if isinstance(original, UpstreamFailure) and public_error.__context__ is original:
         public_error.__context__ = error
         if isinstance(public_error, openai.APIStatusError):

litellm/rust_bridge/host/routes/messages.py

@@ -79,7 +79,12 @@ def map_failure(error: Exception, request: Mapping[str, object], request_provide
             llm_provider=request_provider,
         )
     return failures.map_native_failure(
-        error, str(request["model"]), request_provider, request, optional_str(request.get("api_base"))
+        error,
+        str(request["model"]),
+        request_provider,
+        request,
+        optional_str(request.get("api_base")),
+        preserve_model_prefix=True,
     )
 
 

tests/rust-python-harness/strategies/e2e_parity/__init__.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import Final
 
-from ...shared.reporting.models import SURFACES
+from ...shared.reporting.models import SURFACES, Coverage
 from ...shared.reporting.strategy import (
     CaseDefinition,
     ModuleCaseSpec,
@@ -19,8 +19,10 @@ CASES: Final[tuple[CaseDefinition, ...]] = (
     ),
     CaseDefinition(
         "messages",
-        NotImplementedCaseSpec(
-            reason="Bridge unit tests exist, but no standalone end-to-end parity case is registered."
+        ModuleCaseSpec(
+            coverage=Coverage.PARTIAL,
+            module=f"{__package__}.sdk.messages.case",
+            note="使用本地上游对齐 Anthropic 异步非流式请求、响应和错误，暂不覆盖同步、流式及回调。",
         ),
         surface="sdk",
     ),

tests/rust-python-harness/strategies/e2e_parity/sdk/messages/README.md

@@ -0,0 +1,24 @@
+# Anthropic Messages SDK 对齐
+
+本地回放同一份上游响应，分别执行 Python 和源码编译的 Rust 扩展，比较 HTTP 请求和 SDK 返回值。当前覆盖异步非流式文本、可选参数、400 参数错误及 429 限流错误，暂不覆盖同步、流式和回调
+
+通过公开 SDK `litellm.anthropic_messages` 发起请求，不直接调用底层变换函数。错误用例同时检查兼容层保留上游错误正文和调用者传入的完整模型名，不将原生异常的 `(状态码, 正文)` 元组格式泄露到公开错误消息
+
+先按仓库的锁文件安装依赖并编译扩展，然后运行
+
+```bash
+uv sync --frozen
+LITELLM_LOCAL_MODEL_COST_MAP=True uv run python -m tests.rust-python-harness run e2e_parity --surface sdk --function messages
+```
+
+用例只请求本机回放服务，使用测试凭据，不调用真实模型。缺少编译扩展或候选路径回退到 Python 时，运行失败而非跳过
+
+响应比较保留具体类型和全部公开字段，只排除 `_hidden_params` 内部元数据。成功响应另行检查 `x-litellm-rust` 标记。错误比较使用共享错误模型，包括异常类型、消息、状态码、模型及提供商
+
+用例与防回退校验的测试命令
+
+```bash
+LITELLM_LOCAL_MODEL_COST_MAP=True uv run pytest -o consider_namespace_packages=true tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_case.py -q
+```
+
+`test_python_fallback_cannot_satisfy_messages_parity` 故意将候选路由改为 Python，验证校验会拒绝这种假通过。该策略遵守框架的仅本地运行约束，不新增 CI 工作流

tests/rust-python-harness/strategies/e2e_parity/sdk/messages/case.py

@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import asyncio
+import copy
+import json
+import os
+from collections.abc import Generator, Mapping
+from contextlib import ExitStack, closing, contextmanager
+from functools import partial
+from importlib import import_module
+from typing import Final
+from unittest.mock import patch
+
+from pydantic import TypeAdapter
+
+import litellm
+from litellm.exceptions import BadRequestError, RateLimitError
+from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
+from litellm.messages import dispatch as messages_dispatch
+from litellm.rust_bridge import catalog
+from litellm.rust_bridge.catalog import Route, RouteRule
+from litellm.rust_bridge.configuration import Rollout
+from tests.test_litellm_rust.support.isolation import isolated_callback_registries
+
+from .....shared.parity.compare import assert_request_parity, assert_value_parity
+from .....shared.parity.inprocess import InProcessExecution, run_in_process_async
+from .....shared.parity.models import SDKError, sdk_error_report
+from .....shared.parity.recorded_http import HttpHeader, RecordedHttpResponse
+from .....shared.parity.replay import ReplayServer, replay_server
+from ...runner import E2ECheck
+
+MODEL: Final = "claude-sonnet-5"
+BASELINE_USER_AGENT: Final = "litellm-messages-python-parity"
+RESPONSE_MAPPING: Final = TypeAdapter(Mapping[str, object])
+RESPONSE: Final = {
+    "id": "msg_parity",
+    "type": "message",
+    "role": "assistant",
+    "model": MODEL,
+    "content": [{"type": "text", "text": "Hello from the local provider"}],
+    "stop_reason": "end_turn",
+    "stop_sequence": None,
+    "usage": {"input_tokens": 11, "output_tokens": 7},
+}
+
+
+def recorded_response(body: Mapping[str, object], status: int = 200) -> RecordedHttpResponse:
+    return RecordedHttpResponse.from_bytes(
+        status,
+        (HttpHeader(name="content-type", value="application/json"),),
+        json.dumps(dict(body)).encode(),
+    )
+
+
+async def _invoke(provider_url: str, kwargs: Mapping[str, object]) -> object:
+    try:
+        return await litellm.anthropic_messages(
+            model=f"anthropic/{MODEL}",
+            messages=[{"role": "user", "content": "Hello"}],
+            max_tokens=64,
+            api_key="parity-test-key",
+            api_base=provider_url,
+            timeout=5,
+            num_retries=0,
+            **copy.deepcopy(dict(kwargs)),
+        )
+    except (BadRequestError, RateLimitError) as error:
+        return sdk_error_report(error)
+    finally:
+        await GLOBAL_LOGGING_WORKER.flush()
+
+
+async def _execute(
+    provider: ReplayServer,
+    rollout: Rollout,
+    response: RecordedHttpResponse,
+    kwargs: Mapping[str, object],
+) -> InProcessExecution[object]:
+    rules: Final = (RouteRule(Route.MESSAGES, rollout), *catalog.RULES)
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(catalog, "RULES", rules))
+        if rollout is Rollout.RUST_REQUIRED:
+            stack.enter_context(patch.object(messages_dispatch, "_PYTHON_AMESSAGES", _reject_fallback))
+        return await run_in_process_async(provider, (response,), partial(_invoke, kwargs=kwargs))
+
+
+async def _reject_fallback(*args: object, **kwargs: object) -> object:  # kwargs-ok: 接收公开接口的原始调用参数
+    raise AssertionError("Rust request fell back to Python")
+
+
+def _public_response(response: object) -> object:
+    if isinstance(response, dict):
+        values: Final = RESPONSE_MAPPING.validate_python(response)
+        return {key: value for key, value in values.items() if key != "_hidden_params"}
+    return response
+
+
+async def compare(
+    provider: ReplayServer,
+    response: RecordedHttpResponse,
+    kwargs: Mapping[str, object],
+    expected_error: type[BadRequestError] | type[RateLimitError] | None = None,
+) -> None:
+    baseline: Final = await _execute(provider, Rollout.PYTHON_ONLY, response, kwargs)
+    candidate: Final = await _execute(provider, Rollout.RUST_REQUIRED, response, kwargs)
+
+    assert baseline.requests[0].user_agent == BASELINE_USER_AGENT
+    assert_request_parity(baseline.requests, candidate.requests)
+    assert type(baseline.response) is type(candidate.response)
+    assert_value_parity(_public_response(baseline.response), _public_response(candidate.response))
+    if expected_error is not None:
+        assert isinstance(candidate.response, SDKError)
+        assert candidate.response.exception_type == f"{expected_error.__module__}.{expected_error.__qualname__}"
+        assert candidate.response.status_code == response.status_code
+        return
+    values: Final = RESPONSE_MAPPING.validate_python(candidate.response)
+    hidden: Final = RESPONSE_MAPPING.validate_python(values.get("_hidden_params"))
+    headers: Final = RESPONSE_MAPPING.validate_python(hidden.get("additional_headers"))
+    assert headers.get("x-litellm-rust") == "true"
+
+
+def _check(
+    loop: asyncio.AbstractEventLoop,
+    provider: ReplayServer,
+    response: RecordedHttpResponse,
+    kwargs: Mapping[str, object],
+    expected_error: type[BadRequestError] | type[RateLimitError] | None = None,
+) -> None:
+    loop.run_until_complete(compare(provider, response, kwargs, expected_error))
+
+
+@contextmanager
+def parity_checks() -> Generator[tuple[E2ECheck, ...]]:
+    import_module("litellm.rust_bridge._native")
+
+    with (
+        closing(asyncio.new_event_loop()) as loop,
+        replay_server() as provider,
+        isolated_callback_registries(),
+        patch.object(litellm, "cache", None),  # test-quality-ok: 隔离进程级缓存，确保两次调用均到达上游
+        patch.dict(os.environ, {"LITELLM_USER_AGENT": BASELINE_USER_AGENT}),
+    ):
+        try:
+            yield (
+                E2ECheck("async-text", partial(_check, loop, provider, recorded_response(RESPONSE), {})),
+                E2ECheck(
+                    "async-optional-parameters",
+                    partial(
+                        _check,
+                        loop,
+                        provider,
+                        recorded_response(RESPONSE),
+                        {
+                            "system": "Answer briefly",
+                            "temperature": 1.0,
+                            "stop_sequences": ["done"],
+                            "extra_headers": {"x-parity-test": "messages"},
+                        },
+                    ),
+                ),
+                E2ECheck(
+                    "async-bad-request",
+                    partial(
+                        _check,
+                        loop,
+                        provider,
+                        recorded_response(
+                            {"type": "error", "error": {"type": "invalid_request_error", "message": "bad request"}},
+                            400,
+                        ),
+                        {},
+                        BadRequestError,
+                    ),
+                ),
+                E2ECheck(
+                    "async-rate-limited",
+                    partial(
+                        _check,
+                        loop,
+                        provider,
+                        recorded_response(
+                            {"type": "error", "error": {"type": "rate_limit_error", "message": "rate limited"}},
+                            429,
+                        ),
+                        {},
+                        RateLimitError,
+                    ),
+                ),
+            )
+        finally:
+            loop.run_until_complete(GLOBAL_LOGGING_WORKER.stop())

tests/rust-python-harness/strategies/e2e_parity/sdk/messages/test_case.py

@@ -0,0 +1,48 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Final
+from unittest.mock import patch
+
+import pytest
+
+from .....shared.parity.replay import replay_server
+from .....shared.reporting.models import Coverage, HarnessCase, RunStatus
+from .....shared.reporting.strategy import ModuleCaseSpec
+from ...runner import run_e2e_cases
+from . import case
+
+
+async def test_python_fallback_cannot_satisfy_messages_parity() -> None:
+    route_rule: Final = case.RouteRule
+
+    def python_only(route: case.Route, _rollout: case.Rollout) -> case.RouteRule:
+        return route_rule(route, case.Rollout.PYTHON_ONLY)
+
+    with (
+        replay_server() as provider,
+        patch.dict(os.environ, {"LITELLM_USER_AGENT": case.BASELINE_USER_AGENT}),
+        patch.object(case, "RouteRule", python_only),
+    ):
+        try:
+            with pytest.raises(AssertionError, match="fell back to Python"):
+                await case.compare(provider, case.recorded_response(case.RESPONSE), {})
+        finally:
+            await case.GLOBAL_LOGGING_WORKER.stop()
+
+
+def test_messages_parity_runs_against_the_native_extension() -> None:
+    messages: Final = HarnessCase(
+        strategy_id="e2e_parity",
+        strategy_label="End-to-end parity",
+        sdk_function="messages",
+        surface="sdk",
+        spec=ModuleCaseSpec(coverage=Coverage.PARTIAL, module=case.__name__),
+    )
+
+    code, run = run_e2e_cases((messages,), Path(__file__).parent, lambda _: None)
+
+    assert code == 0, run.failures
+    assert run.results[messages.key].status is RunStatus.PASSED
+    assert len(run.results[messages.key].collected) == 4

tests/unit/rust_bridge/host/routes/test_messages.py

@@ -2,6 +2,7 @@ from types import MappingProxyType
 from typing import Final
 
 import pytest
+from pydantic import TypeAdapter
 
 import litellm
 from litellm.constants import DEFAULT_REQUEST_TIMEOUT_SECONDS
@@ -128,6 +129,22 @@ def test_native_request_rejections_map_to_the_public_400() -> None:
     )
 
 
+@pytest.mark.parametrize("model", ("anthropic/claude-sonnet-5", "claude-sonnet-5"))
+@pytest.mark.parametrize("status", (400, 429))
+def test_upstream_failure_preserves_the_model_the_messages_caller_supplied(model: str, status: int) -> None:
+    class UpstreamFailure(Exception):
+        headers: Final = ()
+
+    upstream: Final = UpstreamFailure(status, '{"error":{"message":"provider rejected request"}}')
+
+    mapped: Final = messages_host.map_failure(upstream, {"model": model}, "anthropic")
+
+    assert isinstance(mapped, (litellm.BadRequestError, litellm.RateLimitError))
+    assert TypeAdapter(str).validate_python(getattr(mapped, "model")) == model
+    assert mapped.status_code == status
+    assert mapped.__context__ is upstream
+
+
 def test_stream_hidden_params_projects_upstream_headers_the_way_the_python_handler_does() -> None:
     hidden: Final = messages_host.stream_hidden_params(
         (("request-id", "req_upstream_123"), ("x-ratelimit-remaining-requests", "41"))

tests/unit/rust_bridge/test_failures.py

@@ -4,6 +4,7 @@ from typing import Final
 import pytest
 
 import litellm
+from litellm.llms.anthropic.common_utils import AnthropicError
 from litellm.rust_bridge import failures
 
 
@@ -12,6 +13,36 @@ class UpstreamRateLimited(Exception):
     message = "rate limited"
 
 
+class NativeUpstreamFailure(Exception):
+    headers: Final = (("retry-after", "7"),)
+
+
+@pytest.mark.parametrize(
+    ("status", "body", "error_type"),
+    (
+        (400, '{"error":{"message":"bad request"}}', litellm.BadRequestError),
+        (429, '{"error":{"message":"rate limited"}}', litellm.RateLimitError),
+    ),
+)
+def test_native_upstream_message_matches_the_provider_error_without_tuple_formatting(
+    status: int, body: str, error_type: type[Exception]
+) -> None:
+    native: Final = NativeUpstreamFailure(status, body)
+    model: Final = "anthropic/claude-sonnet-5"
+    request: Final = MappingProxyType[str, object]({})
+    expected: Final = failures.map_failure(AnthropicError(status, body), model, "anthropic", request)
+
+    mapped: Final = failures.map_native_failure(native, model, "anthropic", request)
+
+    assert type(mapped) is error_type
+    assert str(mapped) == str(expected)
+    assert isinstance(mapped, (litellm.BadRequestError, litellm.RateLimitError))
+    assert mapped.status_code == status
+    assert mapped.response.text == body
+    assert mapped.response.headers["retry-after"] == "7"
+    assert mapped.__context__ is native
+
+
 def test_upstream_status_maps_onto_the_public_exception_contract() -> None:
     upstream: Final = UpstreamRateLimited("rate limited")