fix(batches): stamp the model group on the proxy's model-encoded retrieve path

The model-encoded batch id path calls the SDK directly, so the router never
labels it. Stamp the decoded group into the request's litellm_metadata, and
guard usage-based-routing-v2 the same way the other strategies already are.

litellm/proxy/batches_endpoints/endpoints.py

@@ -52,6 +52,20 @@ from litellm.types.llms.openai import LiteLLMBatchCreateRequest
 router: Final = APIRouter()
 
 
+def _litellm_metadata_of(data: dict) -> dict:
+    """The request's litellm_metadata mapping, created on the request when it carries none.
+
+    The success handler reads this mapping, so a flag or a model group set here has to live
+    inside it rather than beside it.
+    """
+    existing: Final = data.get("litellm_metadata")
+    if isinstance(existing, dict):
+        return existing
+    created: Final = {}  # mutable-ok: the logging layer copies and extends this mapping, so it cannot be a read-only view
+    data["litellm_metadata"] = created
+    return created
+
+
 def _raise_not_found_when_openai_fallback_unservable(
     requested_provider: "str | None",
     data: Mapping[str, object],
@@ -531,11 +545,7 @@ async def retrieve_batch(
 
         poller_owns_accounting: Final = bool(unified_batch_id) and batch_cost_poller_is_active()
         if poller_owns_accounting:
-            litellm_metadata = data.get("litellm_metadata")
-            if not isinstance(litellm_metadata, dict):
-                litellm_metadata = {}  # mutable-ok: the suppression flag must live inside litellm_metadata for the success handler to read it, and this request carried no mapping to extend
-                data["litellm_metadata"] = litellm_metadata
-            litellm_metadata["batch_ignore_default_logging"] = True
+            _litellm_metadata_of(data)["batch_ignore_default_logging"] = True
 
         # Retrieve from provider (for non-terminal states or if DB lookup failed)
         # SCENARIO 1: Batch ID is encoded with model info
@@ -558,6 +568,7 @@ async def retrieve_batch(
             # so litellm.aretrieve_batch can load BedrockBatchesConfig. Without
             # it the call falls into the legacy provider switch and 400s.
             data["model"] = model_from_id
+            _litellm_metadata_of(data).setdefault("model_group", model_from_id)
 
             # Retrieve batch using model credentials
             response = await litellm.aretrieve_batch(

litellm/router_strategy/lowest_tpm_rpm_v2.py

@@ -12,6 +12,7 @@ from litellm._logging import verbose_logger, verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs
+from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
 from litellm.types.router import RouterErrors
 from litellm.types.utils import LiteLLMPydanticObjectBase, StandardLoggingPayload
 from litellm.utils import get_utc_datetime, print_verbose
@@ -210,6 +211,8 @@ class LowestTPMLoggingHandler_v2(BaseRoutingStrategy, CustomLogger):
             return deployment  # don't fail calls if eg. redis fails to connect
 
     def log_success_event(self, kwargs, response_obj, start_time, end_time):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+            return
         try:
             """
             Update TPM/RPM usage on success
@@ -250,6 +253,8 @@ class LowestTPMLoggingHandler_v2(BaseRoutingStrategy, CustomLogger):
             )
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+            return
         try:
             """
             Update TPM usage on success

tests/test_litellm/proxy/batches_endpoints/test_endpoints.py

@@ -1233,9 +1233,11 @@ async def call_retrieve(
     user: Optional[UserAPIKeyAuth] = None,
     headers: Optional[Dict[str, str]] = None,
     query: Optional[Dict[str, str]] = None,
+    enriched_data: Optional[Dict[str, Any]] = None,
 ):
-    # Mirror the real flow: data starts as RetrieveBatchRequest(batch_id=...).
-    harness.data["data"] = {"batch_id": batch_id}
+    # Mirror the real flow: data starts as RetrieveBatchRequest(batch_id=...),
+    # then pre-call enrichment adds key/team metadata to it.
+    harness.data["data"] = {"batch_id": batch_id, **(enriched_data or {})}
     return await endpoints.retrieve_batch(
         request=FakeRequest(headers=headers, query=query),
         fastapi_response=Response(),
@@ -1271,6 +1273,7 @@ async def test_retrieve__model_encoded_id(retrieve_harness):
         "api_key": "sk-azure",
         "api_base": "https://azure.test",
         "model": "azure/gpt-4o",
+        "litellm_metadata": {"model_group": "azure/gpt-4o"},
     }
 
     # 4. OUTPUT SHAPE - ids re-encoded with the model for the round-trip.
@@ -1293,6 +1296,39 @@ async def test_retrieve__model_encoded_id__forwards_decoded_model_not_deployment
     assert retrieve_harness.aretrieve_kwargs()["model"] == "azure/gpt-4o"
 
 
+@pytest.mark.asyncio
+async def test_retrieve__model_encoded_id__stamps_model_group(retrieve_harness):
+    """This path never goes through the router, so nothing else labels the call.
+    Without the stamp the spend log lands under a blank model group and the batch
+    disappears from per-model usage."""
+    await call_retrieve(retrieve_harness, AZURE_BATCH_ID)
+
+    litellm_metadata = retrieve_harness.aretrieve_kwargs()["litellm_metadata"]
+
+    assert litellm_metadata["model_group"] == "azure/gpt-4o"
+
+
+@pytest.mark.asyncio
+async def test_retrieve__model_encoded_id__stamps_model_group_beside_existing_metadata(
+    retrieve_harness,
+):
+    """The stamp joins the metadata pre-call enrichment already built. Replacing
+    that dict instead of adding to it drops the key and team labels the spend log
+    is attributed with."""
+    await call_retrieve(
+        retrieve_harness,
+        AZURE_BATCH_ID,
+        enriched_data={"litellm_metadata": {"user_api_key_alias": "team-a-key"}},
+    )
+
+    litellm_metadata = retrieve_harness.aretrieve_kwargs()["litellm_metadata"]
+
+    assert litellm_metadata == {
+        "user_api_key_alias": "team-a-key",
+        "model_group": "azure/gpt-4o",
+    }
+
+
 @pytest.mark.asyncio
 async def test_retrieve__model_encoded_id__encodes_output_and_error_ids(
     retrieve_harness,

tests/test_litellm/test_router.py

@@ -1169,17 +1169,19 @@ async def test_arouter_aretrieve_batch_does_not_consume_deployment_rate_limits(m
 _ROUTING_STRATEGY_CACHE_MARKERS = ("_map", "_request_count", ":tpm:", ":rpm:")
 
 
-async def _router_strategy_keys(router, timeout: float = 2.0) -> list[str]:
+async def _moved_routing_counters(router, timeout: float = 2.0) -> list[str]:
     loop = asyncio.get_event_loop()
     deadline = loop.time() + timeout
     while loop.time() < deadline:
-        keys = sorted(
-            key
-            for key in router.cache.in_memory_cache.cache_dict
+        cache_dict = router.cache.in_memory_cache.cache_dict
+        moved = sorted(
+            f"{key}={cache_dict[key]}"
+            for key in cache_dict
             if any(marker in key for marker in _ROUTING_STRATEGY_CACHE_MARKERS)
+            and cache_dict[key]
         )
-        if keys:
-            return keys
+        if moved:
+            return moved
         await asyncio.sleep(0.05)
     return []
 
@@ -1212,7 +1214,13 @@ def _batch_fan_out_router(routing_strategy: str):
 
 @pytest.mark.parametrize(
     "routing_strategy",
-    ["usage-based-routing", "latency-based-routing", "cost-based-routing", "least-busy"],
+    [
+        "usage-based-routing",
+        "usage-based-routing-v2",
+        "latency-based-routing",
+        "cost-based-routing",
+        "least-busy",
+    ],
 )
 @pytest.mark.asyncio
 async def test_arouter_aretrieve_batch_does_not_feed_routing_strategies(
@@ -1239,10 +1247,10 @@ async def test_arouter_aretrieve_batch_does_not_feed_routing_strategies(
         for _ in range(3):
             response = await router.aretrieve_batch(batch_id=_BATCH_ID)
         await collector.retrieve_batch_payload()
-        strategy_keys = await _router_strategy_keys(router)
+        moved_counters = await _moved_routing_counters(router)
 
     assert response.id == _BATCH_ID
-    assert strategy_keys == []
+    assert moved_counters == []
 
 
 @pytest.mark.asyncio