diff --git a/litellm/router_strategy/complexity_router/jev_classifier.py b/litellm/router_strategy/complexity_router/jev_classifier.py index ef613f8321d..a41df18b55f 100644 --- a/litellm/router_strategy/complexity_router/jev_classifier.py +++ b/litellm/router_strategy/complexity_router/jev_classifier.py @@ -8,6 +8,7 @@ import httpx from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError import litellm +from litellm._logging import verbose_router_logger from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.litellm_core_utils.internal_call_metadata import ( effective_turn_off_message_logging, @@ -101,7 +102,10 @@ class HttpJevClassifierClient: timeout=timeout_s, ) response.raise_for_status() - self._log_response(request, response, request_kwargs, start_time) + try: + self._log_response(request, response, request_kwargs, start_time) + except Exception as exc: # noqa: BLE001 # logging integrations must not discard a provider verdict + verbose_router_logger.warning("JEV response logging failed (%s)", type(exc).__name__) return TypeAdapter(JevSystemOneResponse).validate_python(response.json()) @staticmethod @@ -163,16 +167,19 @@ class HttpJevClassifierClient: request_body=MappingProxyType({"model": request.model}), litellm_params=params, ) - GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue( - logging_obj.dispatch_success_handlers( - result=normalized["result"], - start_time=start_time, - end_time=end_time, - cache_hit=False, - prefer_async_handlers=True, - **TypeAdapter(dict[str, object]).validate_python(normalized["kwargs"]), - ) + success_handlers: Final = logging_obj.dispatch_success_handlers( + result=normalized["result"], + start_time=start_time, + end_time=end_time, + cache_hit=False, + prefer_async_handlers=True, + **TypeAdapter(dict[str, object]).validate_python(normalized["kwargs"]), ) + try: + GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(success_handlers) + except BaseException: + success_handlers.close() + raise class JevVerdict(NamedTuple): diff --git a/tests/unit/router_strategy/complexity_router/test_jev_classifier.py b/tests/unit/router_strategy/complexity_router/test_jev_classifier.py index dae037ff47c..f7c656cc6cf 100644 --- a/tests/unit/router_strategy/complexity_router/test_jev_classifier.py +++ b/tests/unit/router_strategy/complexity_router/test_jev_classifier.py @@ -2,13 +2,14 @@ import asyncio import json from collections.abc import Mapping from datetime import datetime -from typing import Final +from typing import Final, NoReturn from unittest.mock import create_autospec import httpx import pytest import litellm +from litellm._logging import verbose_router_logger from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER @@ -40,6 +41,66 @@ class _UsageRecorder(CustomLogger): self.calls = (*self.calls, kwargs) +class _UncopyableAuth: + budget_reservation: Final = "parent-reservation" + + def __init__(self, error: Exception) -> None: + self.error = error + + def model_copy(self, *, update: Mapping[str, object]) -> NoReturn: + raise self.error + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("metadata", "error_name"), + [ + ({1: "private-metadata"}, "ValidationError"), + ({"user_api_key_auth": _UncopyableAuth(RuntimeError("private-metadata"))}, "RuntimeError"), + ({"user_api_key_auth": _UncopyableAuth(TimeoutError("private-metadata"))}, "TimeoutError"), + ], +) +async def test_jev_logging_failure_preserves_verdict_and_keeps_circuit_closed( + caplog: pytest.LogCaptureFixture, metadata: Mapping[object, object], error_name: str +) -> None: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + return httpx.Response( + 200, + json={ + "answers": {"tier": _answer().model_dump()}, + "usage": {"input_tokens": 3, "output_tokens": 2}, + }, + ) + + handler: Final = AsyncHTTPHandler() + handler.client = httpx.AsyncClient(transport=httpx.MockTransport(respond)) + router: Final = ComplexityRouter( + "jev-logging-failure", + litellm.Router(model_list=[]), + {"classifier_type": "jev", "jev_classifier_config": {}, "tiers": {"SIMPLE": "cheap"}}, + jev_client=HttpJevClassifierClient("test", "https://typesafe.test", handler), + derive_savings_baseline=False, + ) + with caplog.at_level("WARNING", logger=verbose_router_logger.name): + outcomes: Final = tuple( + [await router.aclassify("choose a tier", request_kwargs={"metadata": metadata}) for _ in range(2)] + ) + await handler.client.aclose() + + assert tuple( + (outcome.cause, outcome.jev_verdict.label if outcome.jev_verdict else None) for outcome in outcomes + ) == ( + ("jev_classifier", "SIMPLE"), + ("jev_classifier", "SIMPLE"), + ) + assert len(requests) == 2 + assert caplog.messages == [f"JEV response logging failed ({error_name})"] * 2 + assert "private-metadata" not in caplog.text + + @pytest.mark.asyncio @pytest.mark.parametrize("status_code", [400, 429, 500, 503]) async def test_jev_http_errors_do_not_dispatch_successful_usage(