fix(proxy): log each endpoint's own call type on failed audio, moderation, image, and rerank spend rows (#49684)

* fix(proxy): record the route's call type on failed speech spend rows

A failed /v1/audio/speech request never puts a logging object on the request
data, so the failure lift returned nothing and the spend row's call_type came
out empty (or whatever the client sent in the body). Without a logging object,
lift the call type the route already resolved, so a failure row records the
same call type a success row on that route records

* fix(proxy): keep the call type of a failure logged under its own route

An MCP tool listing that fails inside a Responses, chat, or messages call
is logged under /mcp/list_tools, so its spend row keeps list_mcp_tools
instead of taking the outer request's call type

* fix(proxy): take a failed row's call type from its endpoint, not a route path guess

* fix(proxy): log the endpoint call type on failed image generation and rerank rows

* fix(proxy): estimate failed-call ledger usage from the route's call type again

The failure usage snapshot now gets the same call type the spend row uses for
its estimate, so a failed Responses call counts input from the newest
compaction item again, as on main.

The endpoint call type tests now go through real routers whose provider
answers 400, instead of asserting on a mocked failure hook.

* test(proxy): type the failed spend row tests and fixture

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>

litellm/proxy/image_endpoints/endpoints.py

@@ -37,6 +37,7 @@ from litellm.proxy.common_utils.openai_error_payload import (
 from litellm.proxy.route_llm_request import route_request
 from litellm.types.images.main import ImageEditRequestParams
 from litellm.types.llms.openai import ChatCompletionUserMessage
+from litellm.types.utils import CallTypes
 
 router: Final = APIRouter()
 _IMAGE_GENERATION_MODEL_ADAPTER: Final[TypeAdapter[object]] = TypeAdapter(object)
@@ -221,7 +222,10 @@ async def image_generation(
         return response
     except Exception as e:
         await proxy_logging_obj.post_call_failure_hook(
-            user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data
+            user_api_key_dict=user_api_key_dict,
+            original_exception=e,
+            request_data=data,
+            dispatched_call_type=CallTypes.aimage_generation.value,
         )
         log_llm_api_exception(e, litellm_call_id)
         if isinstance(e, HTTPException):

litellm/proxy/proxy_server.py

@@ -207,6 +207,7 @@ from litellm.rust_bridge.catalog import Target
 from litellm.types.caching import RedisPipelineIncrementOperation
 from litellm.types.utils import (
     PRICING_OVERRIDES_KEY,
+    CallTypes,
     CountContextManagement,
     LlmProviders,
     ModelResponse,
@@ -13560,7 +13561,10 @@ async def moderations(
         return response
     except Exception as e:
         await proxy_logging_obj.post_call_failure_hook(
-            user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data
+            user_api_key_dict=user_api_key_dict,
+            original_exception=e,
+            request_data=data,
+            dispatched_call_type=CallTypes.amoderation.value,
         )
         log_llm_api_exception(e, litellm_call_id)
         if isinstance(e, ProxyException):
@@ -13705,6 +13709,7 @@ async def audio_speech(
             user_api_key_dict=user_api_key_dict,
             original_exception=e,
             request_data=data,
+            dispatched_call_type=CallTypes.aspeech.value,
         )
         log_llm_api_exception(e, litellm_call_id)
         if isinstance(e, ProxyException):
@@ -13856,7 +13861,10 @@ async def audio_transcriptions(
         return response
     except Exception as e:
         await proxy_logging_obj.post_call_failure_hook(
-            user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data
+            user_api_key_dict=user_api_key_dict,
+            original_exception=e,
+            request_data=data,
+            dispatched_call_type=CallTypes.atranscription.value,
         )
         log_llm_api_exception(e, litellm_call_id)
         if isinstance(e, HTTPException):

litellm/proxy/rerank_endpoints/endpoints.py

@@ -20,6 +20,7 @@ from litellm.proxy.common_utils.openai_error_payload import (
     openai_error_param,
     openai_error_type,
 )
+from litellm.types.utils import CallTypes
 
 router: Final = APIRouter()
 
@@ -114,7 +115,10 @@ async def rerank(
         return response
     except Exception as e:
         await proxy_logging_obj.post_call_failure_hook(
-            user_api_key_dict=user_api_key_dict, original_exception=e, request_data=data
+            user_api_key_dict=user_api_key_dict,
+            original_exception=e,
+            request_data=data,
+            dispatched_call_type=CallTypes.arerank.value,
         )
         log_llm_api_exception(e, litellm_call_id)
         if isinstance(e, HTTPException):

litellm/proxy/utils.py

@@ -1760,7 +1760,10 @@ def _proxy_only_error_log_fields(
 
 
 def _failure_fields_to_lift(
-    request_data: Mapping[str, object], provider_billed: bool = False, dispatched_call_type: str | None = None
+    request_data: Mapping[str, object],
+    provider_billed: bool = False,
+    dispatched_call_type: str | None = None,
+    route_call_type: str | None = None,
 ) -> Mapping[str, object]:
     """Failure-path callbacks run after ``litellm_logging_obj`` is popped from
     request_data (it is not serialisable), so the caller merges these fields
@@ -1770,7 +1773,8 @@ def _failure_fields_to_lift(
     attribution on failed-request spend logs."""
     _logging_obj: Final = request_data.get("litellm_logging_obj")
     if _logging_obj is None:
-        return _EMPTY_LIFT
+        return _EMPTY_LIFT if dispatched_call_type is None else MappingProxyType({"call_type": dispatched_call_type})
+    _estimate_call_type: Final = dispatched_call_type or route_call_type
     _model_call_details: Final = getattr(_logging_obj, "model_call_details", {})
     _first_handoff: Final = _model_call_details.get("first_api_call_start_time")
     _litellm_params: Final = getattr(_logging_obj, "litellm_params", None)
@@ -1785,10 +1789,10 @@ def _failure_fields_to_lift(
         provider_billed=_billed,
         deployment_id=_deployment_id,
         base_model=_base_model,
-        dispatched_call_type=dispatched_call_type,
+        dispatched_call_type=_estimate_call_type,
     )
     _prompt_estimate: Final = (
-        _estimated_prompt_usage(_model_call_details, request_data, dispatched_call_type)
+        _estimated_prompt_usage(_model_call_details, request_data, _estimate_call_type)
         if _billed and _first_handoff is not None and not _model_call_details.get("failed_attempts_usage")
         else None
     )
@@ -1808,7 +1812,7 @@ def _failure_fields_to_lift(
             _billed,
             _deployment_id,
             _base_model,
-            dispatched_call_type,
+            _estimate_call_type,
         )
         if Logging.usage_capture_callback is not None
         and (_billed or _model_call_details.get("combined_usage_object") is not None)
@@ -4104,13 +4108,15 @@ class ProxyLogging:
                 route=route,
                 original_exception=original_exception,
                 error_type=error_type,
+                dispatched_call_type=dispatched_call_type,
             )
 
         request_data.update(
             await offload_token_count(_failure_fields_to_lift)(
                 request_data,
                 provider_billed=_provider_billed_the_failure(original_exception),
-                dispatched_call_type=dispatched_call_type or _call_type_for_route(user_api_key_dict.request_route),
+                dispatched_call_type=dispatched_call_type,
+                route_call_type=_call_type_for_route(user_api_key_dict.request_route),
             )
         )
 
@@ -4199,6 +4205,7 @@ class ProxyLogging:
         route: str | None = None,
         original_exception: Exception | None = None,
         error_type: ProxyErrorTypes | None = None,
+        dispatched_call_type: str | None = None,
     ):
         """
         Handle logging for proxy only errors by calling `litellm_logging_obj.async_failure_handler`
@@ -4298,7 +4305,7 @@ class ProxyLogging:
                 litellm_logging_obj.model_call_details["input"] = input
                 body_shape_call_type = CallTypes.aembedding.value
             resolved_call_type: Final = _call_type_for_route(route) or (
-                None if set_up_by_request else body_shape_call_type
+                None if set_up_by_request else dispatched_call_type or body_shape_call_type
             )
             if resolved_call_type is not None and litellm_logging_obj.call_type != CallTypes.pass_through.value:
                 litellm_logging_obj.call_type = resolved_call_type

tests/unit/proxy/image_endpoints/test_endpoints.py

@@ -715,3 +715,29 @@ def test_image_generation_uses_key_num_retries(monkeypatch: pytest.MonkeyPatch)
     router: Final = proxy_server.llm_router
     assert router is not None
     assert get_deployment_failures_for_current_minute(router, "image-regression-failing") == 3
+
+
+def test_failed_image_generation_spend_row_records_aimage_generation(monkeypatch: pytest.MonkeyPatch) -> None:
+    from unittest.mock import AsyncMock, MagicMock
+
+    from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+    from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+
+    spend_writer: Final = MagicMock(update_database=AsyncMock())
+    monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+    client, handler = _image_generation_client(monkeypatch, UserAPIKeyAuth(api_key="sk-test"))
+    try:
+        with client:
+            response: Final = client.post("/v1/images/generations", json={"model": "A", "prompt": "draw a tree"})
+    finally:
+        ProxyLogging._callback_capabilities_cache.clear()
+
+    assert (response.status_code, handler.calls) == (429, ["failing"]), response.text
+    db_call: Final = spend_writer.update_database.call_args.kwargs
+    payload: Final = get_logging_payload(
+        kwargs=db_call["kwargs"],
+        response_obj=db_call["completion_response"],
+        start_time=db_call["start_time"],
+        end_time=db_call["end_time"],
+    )
+    assert (payload["status"], payload["call_type"]) == ("failure", "aimage_generation")

tests/unit/proxy/proxy_server/conftest.py

@@ -17,6 +17,8 @@ from unittest.mock import AsyncMock, MagicMock
 import pytest
 
 if TYPE_CHECKING:
+    import respx
+
     from litellm import Router
 
 from tests.unit.proxy.router_entry_point_stubs import route_entry_points_to_stubs
@@ -563,3 +565,72 @@ def openai_only_router(monkeypatch: pytest.MonkeyPatch) -> Router:
     monkeypatch.setattr(proxy_server, "general_settings", {})
     monkeypatch.setattr(proxy_server, "user_model", None)
     return router
+
+
+@pytest.fixture
+def failed_spend_row(
+    monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
+) -> Iterator[Callable[[], tuple[str, str]]]:
+    """Route tts-1, whisper-1 and omni-moderation to an OpenAI deployment whose provider answers 400,
+    and return a reader for the (status, call_type) of the spend row the failure wrote."""
+    import litellm
+    from litellm import Router
+    from litellm.proxy import proxy_server
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+    from litellm.proxy.proxy_server import ProxyConfig
+    from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+    from litellm.proxy.utils import ProxyLogging
+
+    api_base: Final = "https://failing-provider.example/v1"
+    respx_mock.post(url__startswith=api_base).respond(
+        400, json={"error": {"message": "rejected by the provider", "type": "invalid_request_error"}}
+    )
+    router: Final = Router(
+        model_list=[
+            {
+                "model_name": model_name,
+                "litellm_params": {
+                    "model": f"openai/{provider_model}",
+                    "api_base": api_base,
+                    "api_key": "provider-key",
+                    "max_retries": 0,
+                },
+            }
+            for model_name, provider_model in (
+                ("tts-1", "tts-1"),
+                ("whisper-1", "whisper-1"),
+                ("omni-moderation", "omni-moderation-latest"),
+            )
+        ],
+        num_retries=0,
+    )
+    cache: Final = UserApiKeyCache()
+    spend_writer: Final = MagicMock(update_database=AsyncMock())
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+    for name, value in (
+        ("llm_router", router),
+        ("llm_model_list", router.model_list),
+        ("prisma_client", None),
+        ("general_settings", {}),
+        ("user_model", None),
+        ("proxy_config", ProxyConfig()),
+        ("proxy_logging_obj", ProxyLogging(user_api_key_cache=cache)),
+        ("user_api_key_cache", cache),
+        ("version", "test-version"),
+    ):
+        monkeypatch.setattr(proxy_server, name, value)
+
+    def last_row() -> tuple[str, str]:
+        db_call: Final = spend_writer.update_database.call_args.kwargs
+        payload: Final = get_logging_payload(
+            kwargs=db_call["kwargs"],
+            response_obj=db_call["completion_response"],
+            start_time=db_call["start_time"],
+            end_time=db_call["end_time"],
+        )
+        return payload["status"], payload["call_type"]
+
+    yield last_row
+    ProxyLogging._callback_capabilities_cache.clear()

tests/unit/proxy/proxy_server/test_routes_audio.py

@@ -9,7 +9,9 @@ Pins (PR2):
 
 from __future__ import annotations
 
+import contextlib
 import io
+from typing import TYPE_CHECKING, Callable, Final
 from unittest.mock import AsyncMock, MagicMock
 
 import httpx
@@ -22,6 +24,9 @@ from litellm.proxy._types import ProxyErrorTypes, ProxyException
 from litellm.types.llms.openai import HttpxBinaryResponseContent
 from litellm.types.router import RouterErrors
 
+if TYPE_CHECKING:
+    from fastapi.testclient import TestClient
+
 _TAG_ROUTING_DENIAL = f"{RouterErrors.no_deployments_with_tag_routing.value}. Passed model=tts-1 and tags=['x']"
 
 
@@ -359,3 +364,32 @@ def test_audio_transcription_error(client, auth_as, patched_transcription_error,
         response = client.post(path, files=files, data=data)
     assert response.status_code == 500
     assert len(response.content) > 0
+
+
+def _post_speech(client: TestClient) -> httpx.Response:
+    return client.post("/v1/audio/speech", json={"model": "tts-1", "input": "Hi", "voice": "alloy"})
+
+
+def _post_transcription(client: TestClient) -> httpx.Response:
+    return client.post(
+        "/v1/audio/transcriptions",
+        files={"file": ("audio.mp3", b"\x00\x01\x02", "audio/mpeg")},
+        data={"model": "whisper-1"},
+    )
+
+
+@pytest.mark.parametrize(
+    ("send", "call_type"),
+    [(_post_speech, "aspeech"), (_post_transcription, "atranscription")],
+)
+def test_a_provider_rejection_logs_the_audio_endpoint_call_type(
+    client: TestClient,
+    auth_as: Callable[..., contextlib.AbstractContextManager[object]],
+    failed_spend_row: Callable[[], tuple[str, str]],
+    send: Callable[[TestClient], httpx.Response],
+    call_type: str,
+) -> None:
+    with auth_as():
+        response: Final = send(client)
+    assert response.status_code == 400, response.text
+    assert failed_spend_row() == ("failure", call_type)

tests/unit/proxy/proxy_server/test_routes_moderations.py

@@ -7,12 +7,17 @@ Pins (PR2):
 
 from __future__ import annotations
 
+import contextlib
+from typing import TYPE_CHECKING, Callable, Final
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
 from litellm.proxy import proxy_server
 
+if TYPE_CHECKING:
+    from fastapi.testclient import TestClient
+
 from .conftest import normalize  # type: ignore[import-not-found]
 
 HAPPY_RESPONSE = {
@@ -109,3 +114,14 @@ def test_moderation_error(client, auth_as, moderation_pipeline_raises, path):
         response = client.post(path, json=payload)
     assert response.status_code == 500
     assert len(response.content) > 0
+
+
+def test_a_provider_rejection_logs_the_moderation_call_type(
+    client: TestClient,
+    auth_as: Callable[..., contextlib.AbstractContextManager[object]],
+    failed_spend_row: Callable[[], tuple[str, str]],
+) -> None:
+    with auth_as():
+        response: Final = client.post("/v1/moderations", json={"model": "omni-moderation", "input": ["one", "two"]})
+    assert response.status_code == 400, response.text
+    assert failed_spend_row() == ("failure", "amoderation")

tests/unit/proxy/rerank_endpoints/test_endpoints.py

@@ -5,10 +5,13 @@ Tests for rerank_endpoints/endpoints.py response headers.
 import json
 import logging
 from collections.abc import Iterator
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-from fastapi import HTTPException, Request, Response
+import respx
+from fastapi import FastAPI, HTTPException, Request, Response
+from fastapi.testclient import TestClient
 
 import litellm.proxy.common_request_processing as common_request_processing_mod
 import litellm.proxy.proxy_server as proxy_server_mod
@@ -209,3 +212,68 @@ async def test_a_rejection_raised_before_routing_keeps_its_own_status(monkeypatc
     error = await _rerank_failure(rejection, raised_before_routing=True, monkeypatch=monkeypatch)
 
     assert (error.type, error.param, error.code) == ("bad_request_error", "session_id", "400")
+
+
+def test_failed_rerank_spend_row_records_arerank(
+    monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
+) -> None:
+    import litellm
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+    from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+    from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+    from litellm.proxy.proxy_server import ProxyConfig
+    from litellm.proxy.rerank_endpoints.endpoints import router as rerank_router
+    from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+    from litellm.proxy.utils import ProxyLogging
+
+    provider: Final = respx_mock.post(url__startswith="https://rerank-provider.example").respond(
+        400, json={"message": "invalid documents"}
+    )
+    cache: Final = UserApiKeyCache()
+    spend_writer: Final = MagicMock(update_database=AsyncMock())
+    monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
+    monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+    monkeypatch.setattr(proxy_server_mod, "general_settings", {})
+    monkeypatch.setattr(
+        proxy_server_mod,
+        "llm_router",
+        litellm.Router(
+            model_list=[
+                {
+                    "model_name": "rerank-model",
+                    "litellm_params": {
+                        "model": "cohere/rerank-v3.5",
+                        "api_base": "https://rerank-provider.example",
+                        "api_key": "provider-key",
+                    },
+                }
+            ],
+            num_retries=0,
+        ),
+    )
+    monkeypatch.setattr(proxy_server_mod, "proxy_config", ProxyConfig())
+    monkeypatch.setattr(proxy_server_mod, "proxy_logging_obj", ProxyLogging(user_api_key_cache=cache))
+    monkeypatch.setattr(proxy_server_mod, "user_api_key_cache", cache)
+    monkeypatch.setattr(proxy_server_mod, "user_model", None)
+    monkeypatch.setattr(proxy_server_mod, "version", "1.2.3")
+    app: Final = FastAPI()
+    app.include_router(rerank_router)
+    app.add_exception_handler(ProxyException, proxy_server_mod.openai_exception_handler)
+    app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(api_key="sk-test")
+    try:
+        with TestClient(app, raise_server_exceptions=False) as client:
+            response: Final = client.post(
+                "/v1/rerank", json={"model": "rerank-model", "query": "q", "documents": ["a", "b"]}
+            )
+    finally:
+        ProxyLogging._callback_capabilities_cache.clear()
+
+    assert (response.status_code, provider.call_count) == (400, 1), response.text
+    db_call: Final = spend_writer.update_database.call_args.kwargs
+    payload: Final = get_logging_payload(
+        kwargs=db_call["kwargs"],
+        response_obj=db_call["completion_response"],
+        start_time=db_call["start_time"],
+        end_time=db_call["end_time"],
+    )
+    assert (payload["status"], payload["call_type"]) == ("failure", "arerank")

tests/unit/proxy/test_proxy_utils_model_creation_and_error_logging.py

@@ -818,6 +818,165 @@ class TestPostCallFailureHookLiftsCallTypeAndStartTime:
         assert payload["model"] == "MCP: deepwiki-ask_question"
         assert payload["endTime"] - payload["startTime"] >= real_datetime.timedelta(seconds=2)
 
+    @pytest.mark.asyncio
+    async def test_failed_speech_spend_row_records_the_endpoint_call_type_over_the_body(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        from types import SimpleNamespace
+
+        from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+        from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+
+        rejected: Final = litellm.BadRequestError(
+            message="Voice name nonexistent is not supported", model="gemini-tts", llm_provider="gemini"
+        )
+        request_data: Final = {
+            "model": "gemini-tts",
+            "input": "Hello there",
+            "voice": "nonexistent",
+            "call_type": "acompletion",
+        }
+        proxy_logging_obj: Final = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = []
+        spend_writer: Final = SimpleNamespace(update_database=AsyncMock())
+        monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+        try:
+            with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()):
+                await proxy_logging_obj.post_call_failure_hook(
+                    request_data=request_data,
+                    original_exception=rejected,
+                    user_api_key_dict=UserAPIKeyAuth(api_key="sk-test", request_route="/v1/audio/speech"),
+                    dispatched_call_type="aspeech",
+                )
+        finally:
+            ProxyLogging._callback_capabilities_cache.clear()
+
+        db_call: Final = spend_writer.update_database.call_args.kwargs
+        payload: Final = get_logging_payload(
+            kwargs=db_call["kwargs"],
+            response_obj=db_call["completion_response"],
+            start_time=db_call["start_time"],
+            end_time=db_call["end_time"],
+        )
+        assert payload["call_type"] == "aspeech"
+
+    @pytest.mark.asyncio
+    async def test_failed_mcp_list_tools_row_inside_a_responses_request_keeps_list_mcp_tools(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        from types import SimpleNamespace
+
+        from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+        from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+
+        request_data: Final = {
+            "model": "MCP: list_tools",
+            "call_type": "list_mcp_tools",
+            "input": [{"role": "system", "content": "list_tools"}],
+        }
+        proxy_logging_obj: Final = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = []
+        spend_writer: Final = SimpleNamespace(update_database=AsyncMock())
+        monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+        try:
+            with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()):
+                await proxy_logging_obj.post_call_failure_hook(
+                    request_data=request_data,
+                    original_exception=RuntimeError("MCP server unreachable"),
+                    user_api_key_dict=UserAPIKeyAuth(api_key="sk-test", request_route="/v1/responses"),
+                    route="/mcp/list_tools",
+                )
+        finally:
+            ProxyLogging._callback_capabilities_cache.clear()
+
+        db_call: Final = spend_writer.update_database.call_args.kwargs
+        payload: Final = get_logging_payload(
+            kwargs=db_call["kwargs"],
+            response_obj=db_call["completion_response"],
+            start_time=db_call["start_time"],
+            end_time=db_call["end_time"],
+        )
+        assert payload["call_type"] == "list_mcp_tools"
+
+    @pytest.mark.asyncio
+    async def test_failure_without_an_endpoint_call_type_does_not_guess_one_from_a_multi_method_path(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        from types import SimpleNamespace
+
+        from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+        from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+
+        proxy_logging_obj: Final = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = []
+        spend_writer: Final = SimpleNamespace(update_database=AsyncMock())
+        monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+        try:
+            with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()):
+                await proxy_logging_obj.post_call_failure_hook(
+                    request_data={},
+                    original_exception=RuntimeError("vector store lookup failed"),
+                    user_api_key_dict=UserAPIKeyAuth(api_key="sk-test", request_route="/v1/vector_stores"),
+                )
+        finally:
+            ProxyLogging._callback_capabilities_cache.clear()
+
+        db_call: Final = spend_writer.update_database.call_args.kwargs
+        payload: Final = get_logging_payload(
+            kwargs=db_call["kwargs"],
+            response_obj=db_call["completion_response"],
+            start_time=db_call["start_time"],
+            end_time=db_call["end_time"],
+        )
+        assert payload["call_type"] == ""
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("request_data", "route", "endpoint_call_type"),
+        [
+            ({"model": "gemini-tts", "input": "Hello there", "voice": "Kore"}, "/v1/audio/speech", "aspeech"),
+            ({"model": "omni-moderation", "input": ["one", "two"]}, "/v1/moderations", "amoderation"),
+            ({"model": "gpt-transcribe"}, "/v1/audio/transcriptions", "atranscription"),
+        ],
+    )
+    async def test_request_rejected_by_the_proxy_records_the_endpoint_call_type(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        request_data: dict[str, object],
+        route: str,
+        endpoint_call_type: str,
+    ):
+        from types import SimpleNamespace
+
+        from fastapi import HTTPException
+
+        from litellm.proxy.hooks.proxy_track_cost_callback import ProxyDBLogger
+        from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
+
+        proxy_logging_obj: Final = ProxyLogging(user_api_key_cache=DualCache())
+        proxy_logging_obj.alert_types = []
+        spend_writer: Final = SimpleNamespace(update_database=AsyncMock())
+        monkeypatch.setattr(litellm, "callbacks", [ProxyDBLogger(spend_writer=lambda: spend_writer)])
+        try:
+            with patch.object(proxy_logging_obj, "update_request_status", new=AsyncMock()):
+                await proxy_logging_obj.post_call_failure_hook(
+                    request_data=dict(request_data),
+                    original_exception=HTTPException(status_code=429, detail="Rate limit exceeded"),
+                    user_api_key_dict=UserAPIKeyAuth(api_key="sk-test", request_route=route),
+                    dispatched_call_type=endpoint_call_type,
+                )
+        finally:
+            ProxyLogging._callback_capabilities_cache.clear()
+
+        db_call: Final = spend_writer.update_database.call_args.kwargs
+        payload: Final = get_logging_payload(
+            kwargs=db_call["kwargs"],
+            response_obj=db_call["completion_response"],
+            start_time=db_call["start_time"],
+            end_time=db_call["end_time"],
+        )
+        assert payload["call_type"] == endpoint_call_type
+
 
 class TestPostCallFailureHookEstimatesDispatchedInputTokens:
     """A non-stream request that failed after dispatch (timeout, provider
@@ -3576,6 +3735,44 @@ def test_an_earlier_timed_out_responses_attempt_counts_its_input_from_the_newest
     assert sent == earlier_attempt_prompt_tokens([COMPACTION_INSTRUCTIONS, COMPACTION_ITEM, TURN_AFTER_COMPACTION])
 
 
+@pytest.mark.asyncio
+async def test_a_billed_responses_route_failure_captures_usage_from_the_newest_compaction_item() -> None:
+    from litellm.litellm_core_utils.get_litellm_params import get_litellm_params
+    from litellm.litellm_core_utils.litellm_logging import UsageCaptureObservation
+    from litellm.proxy.db.accounting_ledger import UsageWrite
+    from litellm.proxy.spend_tracking.usage_capture import UsageCaptureRuntime
+    from litellm.proxy.spend_tracking.usage_meter import UsageCaptureSettings
+    from litellm.proxy.utils import _failure_fields_to_lift
+
+    async def persist(_: tuple[UsageWrite, ...]) -> None:
+        return None
+
+    def captured_prompt_tokens(request_input: list[dict[str, str]]) -> int:
+        logging_obj: Final = _dispatched_failure_logging_obj(FAILURE_USAGE_MODEL, get_litellm_params())
+        logging_obj.model_call_details["call_type"] = "aresponses"
+        logging_obj.model_call_details["messages"] = request_input
+        lifted: Final = _failure_fields_to_lift(
+            {"litellm_logging_obj": logging_obj}, provider_billed=True, route_call_type="aresponses"
+        )
+        snapshot: Final = lifted["_accounting_usage_snapshot"]
+        assert isinstance(snapshot, UsageCaptureObservation) and snapshot.usage_json is not None
+        return json.loads(snapshot.usage_json)["prompt_tokens"]
+
+    runtime: Final = UsageCaptureRuntime()
+    runtime.start(persist, UsageCaptureSettings())
+    try:
+        sent: Final = captured_prompt_tokens(
+            [COMPACTION_INSTRUCTIONS, *COMPACTED_HISTORY, COMPACTION_ITEM, TURN_AFTER_COMPACTION]
+        )
+        since_compaction: Final = captured_prompt_tokens(
+            [COMPACTION_INSTRUCTIONS, COMPACTION_ITEM, TURN_AFTER_COMPACTION]
+        )
+    finally:
+        await runtime.stop(1)
+
+    assert sent == since_compaction
+
+
 @pytest.mark.parametrize(
     "current_attempt_api_call_start_time",
     [None, 1700000001.0],