refactor(vertex-live): type the usage helpers without Any

The two helpers this branch adds took Sequence[Mapping[str, Any]], which the
repo forbids, and only typechecked because Any is compatible with everything.
Both now take Mapping[str, object] and the raw *TokensDetails value is narrowed
to its mapping entries at each of the three call sites.

TypedDicts are the wrong tool here: _merged_modality_totals reads count_key and
details_key as runtime strings, and the aggregation deliberately passes unknown
keys straight through, so both need a mapping whose keys are not literals.

The narrowing is not cosmetic. The handler's only failure path returns no result
at all, so a *TokensDetails value that was not a list of objects used to raise
while being read and cost the whole session its bill.

litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_ai_live_passthrough_logging_handler.py

@@ -9,7 +9,7 @@ from collections.abc import Mapping, Sequence
 from datetime import datetime
 from itertools import chain
 from types import MappingProxyType
-from typing import Any, Final
+from typing import Final
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy.pass_through_endpoints.llm_provider_handlers.base_passthrough_logging_handler import (
@@ -37,6 +37,11 @@ _AGGREGATED_FIELDS: Final = frozenset(
 )
 
 
+def _detail_entries(raw: object) -> tuple[Mapping[str, object], ...]:
+    """Narrow one turn's ``*TokensDetails`` value to the entries that are actually shaped like one."""
+    return tuple(entry for entry in raw if isinstance(entry, Mapping)) if isinstance(raw, Sequence) else ()
+
+
 class VertexAILivePassthroughLoggingHandler(BasePassthroughLoggingHandler):
     """
     Handles cost tracking and logging for Vertex AI Live API WebSocket passthrough.
@@ -68,7 +73,7 @@ class VertexAILivePassthroughLoggingHandler(BasePassthroughLoggingHandler):
 
     @staticmethod
     def _resolve_detail_counts(
-        details: Sequence[Mapping[str, Any]],
+        details: Sequence[Mapping[str, object]],
         declared_total: object,
     ) -> tuple[tuple[str, int], ...]:
         """
@@ -100,7 +105,7 @@ class VertexAILivePassthroughLoggingHandler(BasePassthroughLoggingHandler):
 
     @staticmethod
     def _merged_modality_totals(
-        snapshots: Sequence[Mapping[str, Any]],
+        snapshots: Sequence[Mapping[str, object]],
         count_key: str,
         details_key: str,
     ) -> Mapping[str, int]:
@@ -109,7 +114,7 @@ class VertexAILivePassthroughLoggingHandler(BasePassthroughLoggingHandler):
             tuple(
                 chain.from_iterable(
                     VertexAILivePassthroughLoggingHandler._resolve_detail_counts(
-                        snapshot.get(details_key) or [], snapshot.get(count_key)
+                        _detail_entries(snapshot.get(details_key)), snapshot.get(count_key)
                     )
                     for snapshot in snapshots
                 )
@@ -179,12 +184,13 @@ class VertexAILivePassthroughLoggingHandler(BasePassthroughLoggingHandler):
         """
         prompt_by_modality: Final = VertexAILivePassthroughLoggingHandler._sum_by_modality(
             VertexAILivePassthroughLoggingHandler._resolve_detail_counts(
-                usage_metadata.get("promptTokensDetails") or [], usage_metadata.get("promptTokenCount")
+                _detail_entries(usage_metadata.get("promptTokensDetails")), usage_metadata.get("promptTokenCount")
             )
         )
         candidates_by_modality: Final = VertexAILivePassthroughLoggingHandler._sum_by_modality(
             VertexAILivePassthroughLoggingHandler._resolve_detail_counts(
-                usage_metadata.get("candidatesTokensDetails") or [], usage_metadata.get("candidatesTokenCount")
+                _detail_entries(usage_metadata.get("candidatesTokensDetails")),
+                usage_metadata.get("candidatesTokenCount"),
             )
         )
 

tests/pass_through_unit_tests/test_vertex_ai_live_passthrough.py

@@ -381,6 +381,36 @@ class TestVertexAILivePassthroughLoggingHandler:
         grounded_cost = self._session_cost(handler, mock_logging_obj, grounded, self.NATIVE_AUDIO_MODEL)
         assert grounded_cost == pytest.approx(plain_cost, rel=1e-9), "reporting tool use must not move the bill"
 
+    def test_a_malformed_details_entry_does_not_cost_the_whole_session(self, handler, mock_logging_obj):
+        """A ``*TokensDetails`` value that is not a list of objects must not take the session down.
+
+        The handler's only error path returns no result at all, so one odd frame used to throw
+        while reading it and the whole session billed nothing. The good turns still bill.
+        """
+        turns = self.AUDIO_SESSION[:3]
+        messages = self._live_messages(turns)
+        mangled = [dict(message) for message in messages]
+        mangled[1]["usageMetadata"] = {**mangled[1]["usageMetadata"], "promptTokensDetails": "TEXT"}
+
+        usage = self._session_usage(handler, mock_logging_obj, mangled, self.NATIVE_AUDIO_MODEL)
+
+        surviving = turns[1:]
+        assert usage.prompt_tokens_details.audio_tokens == sum(turn["prompt"][1] for turn in surviving)
+        assert usage.prompt_tokens_details.text_tokens == sum(turn["prompt"][0] for turn in surviving)
+        assert usage.prompt_tokens == sum(sum(turn["prompt"]) for turn in turns), "the totals still cover every turn"
+
+        direct = handler._create_usage_object_from_metadata(
+            usage_metadata={
+                "promptTokenCount": 40,
+                "candidatesTokenCount": 12,
+                "promptTokensDetails": [{"modality": "AUDIO", "tokenCount": 40}, "AUDIO"],
+                "candidatesTokensDetails": {"modality": "TEXT", "tokenCount": 12},
+            },
+            model=self.NATIVE_AUDIO_MODEL,
+        )
+        assert direct.prompt_tokens_details.audio_tokens == 40, "the well-formed entry beside a bad one still counts"
+        assert direct.completion_tokens == 12
+
     @pytest.mark.parametrize(
         "label,prompt_details,candidate_details",
         [