fix(auto-router): isolate JEV verdicts from logging failures

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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):

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(