fix(batches): skip batch line-item events in ClickHouse spend sink and router quota counters (#44486)

* fix(batches): skip batch line-item events in ClickHouse spend sink and router quota counters

Batch line-item callbacks (call_type=acompletion + litellm_params.batch_parent_id)
bypassed the call-type-only batch guards:

- ClickHouseSpendLogger wrote one spend row per line on top of the aggregate
  aretrieve_batch row, billing every batch twice with the flag on.
- Router deployment callbacks and the routing-strategy usage handlers charged the
  historical batch's tokens/rpm to the live TPM/RPM counters, so after
  Router.aretrieve_batch the next chat raised RouterRateLimitError.

Skip these events wherever the aggregate batch retrieve is already skipped.

* fix(logging): keep internal litellm_params authoritative in model_call_details

A request body key named litellm_params lands in optional_params, and
update_environment_variables merged **optional_params after the internal
entry, so a caller could overwrite the logging kwargs' litellm_params and
forge markers like batch_parent_id. Live-reproduced: a Gemini chat whose
body carried {"litellm_params": {"batch_parent_id": ...}} returned 200
with no ClickHouse spend row. Re-assert the internal litellm_params after
the merge so every sink (ClickHouse, Postgres, router quota counters)
reads the same trusted value, and assert the aggregate aretrieve_batch
row still writes in the ClickHouse line-item test (review finding).

litellm/integrations/clickhouse/clickhouse_spend_logger.py

@@ -16,6 +16,7 @@ from litellm._logging import verbose_logger
 from litellm.integrations.clickhouse.clickhouse_batch_logger import ClickHouseBatchLogger
 from litellm.integrations.clickhouse.context import is_lens_analysis
 from litellm.integrations.clickhouse.schema import SPEND_LOGS_TABLE
+from litellm.litellm_core_utils.core_helpers import is_batch_line_item_event
 from litellm.tracing.types import SpendLogRecord
 from litellm.types.utils import StandardLoggingPayload
 
@@ -154,6 +155,10 @@ class ClickHouseSpendLogger(ClickHouseBatchLogger):
     table = SPEND_LOGS_TABLE
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time) -> None:
+        # Batch line items are billed by the aggregate aretrieve_batch row; a per-line
+        # spend row here would bill the batch twice.
+        if is_batch_line_item_event(kwargs):
+            return
         self._log(kwargs)
 
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time) -> None:

litellm/litellm_core_utils/litellm_logging.py

@@ -947,6 +947,10 @@ class Logging(LiteLLMLoggingBaseClass):
                 **additional_params,
             }
         )
+        # Provider params / additional kwargs are caller-influenced; they must not
+        # overwrite the internal litellm_params, or a request body carrying
+        # `litellm_params` could forge logging markers (e.g. batch_parent_id).
+        self.model_call_details["litellm_params"] = self.litellm_params
 
         ## check if stream options is set ##  - used by CustomStreamWrapper for easy instrumentation
         if "stream_options" in additional_params:

litellm/router.py

@@ -78,6 +78,7 @@ from litellm.litellm_core_utils.core_helpers import (
     get_litellm_metadata_from_kwargs,
     get_metadata_variable_name_from_kwargs,
     get_or_create_metadata_bucket,
+    is_batch_line_item_event,
 )
 from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
 from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
@@ -8050,7 +8051,7 @@ class Router:
             # WS session wrappers fire with result=None; per-turn costs tracked by inner calls.
             if kwargs.get("call_type") in ("_aresponses_websocket", "_arealtime"):
                 return
-            if is_batch_retrieve_call_type(kwargs.get("call_type")):
+            if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
                 return
             standard_logging_object: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object", None)
             if standard_logging_object is None:
@@ -8198,7 +8199,7 @@ class Router:
         - key: str - The key used to increment the cache
         - None: if no key is found
         """
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return None
         id = None
         if kwargs["litellm_params"].get("metadata") is None:
@@ -8328,7 +8329,7 @@ class Router:
         """
         Update RPM usage for a deployment
         """
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         deployment_name: Final = kwargs["litellm_params"]["metadata"].get(
             "deployment", None

litellm/router_strategy/least_busy.py

@@ -9,6 +9,7 @@ from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.caching.redis_cache import log_redis_failure
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.core_helpers import is_batch_line_item_event
 from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
 
 IN_FLIGHT_COUNT_TTL_SECONDS: Final = 60 * 60
@@ -49,7 +50,7 @@ def _request_count_key(model_group: str, deployment_id: str) -> str:
 
 
 def _deployment_ref(kwargs: Mapping[str, object]) -> tuple[str, str] | None:
-    if is_batch_retrieve_call_type(kwargs.get("call_type")):
+    if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
         return None
     try:
         call: Final = _CALL_KWARGS.validate_python(kwargs)

litellm/router_strategy/lowest_cost.py

@@ -8,6 +8,7 @@ from litellm import ModelResponse, token_counter, verbose_logger
 from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.core_helpers import is_batch_line_item_event
 from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
 
 
@@ -20,7 +21,7 @@ class LowestCostLoggingHandler(CustomLogger):
         self.router_cache = router_cache
 
     def log_success_event(self, kwargs, response_obj, start_time, end_time):
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """
@@ -95,7 +96,7 @@ class LowestCostLoggingHandler(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")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """

litellm/router_strategy/lowest_latency.py

@@ -12,7 +12,11 @@ import litellm
 from litellm import ModelResponse, token_counter, verbose_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, safe_divide_seconds
+from litellm.litellm_core_utils.core_helpers import (
+    _get_parent_otel_span_from_kwargs,
+    is_batch_line_item_event,
+    safe_divide_seconds,
+)
 from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
 from litellm.types.utils import LiteLLMPydanticObjectBase
 
@@ -59,7 +63,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
         self.routing_args = RoutingArgs(**routing_args)
 
     def log_success_event(self, kwargs, response_obj, start_time, end_time):
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """
@@ -185,7 +189,7 @@ class LowestLatencyLoggingHandler(CustomLogger):
         """
         Check if Timeout Error, if timeout set deployment latency -> 100
         """
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             metadata_field: Final = self._select_metadata_field(kwargs)
@@ -241,7 +245,7 @@ class LowestLatencyLoggingHandler(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")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """

litellm/router_strategy/lowest_tpm_rpm.py

@@ -8,6 +8,7 @@ from litellm import token_counter
 from litellm._logging import verbose_router_logger
 from litellm.caching.caching import DualCache
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.litellm_core_utils.core_helpers import is_batch_line_item_event
 from litellm.router_utils.batch_utils import is_batch_retrieve_call_type
 from litellm.types.utils import LiteLLMPydanticObjectBase
 from litellm.utils import print_verbose
@@ -28,7 +29,7 @@ class LowestTPMLoggingHandler(CustomLogger):
         self.routing_args = RoutingArgs(**routing_args)
 
     def log_success_event(self, kwargs, response_obj, start_time, end_time):
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """
@@ -82,7 +83,7 @@ class LowestTPMLoggingHandler(CustomLogger):
             verbose_router_logger.debug(traceback.format_exc())
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
-        if is_batch_retrieve_call_type(kwargs.get("call_type")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """

litellm/router_strategy/lowest_tpm_rpm_v2.py

@@ -14,7 +14,7 @@ from litellm import token_counter
 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.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs, is_batch_line_item_event
 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
@@ -250,7 +250,7 @@ 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")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """
@@ -292,7 +292,7 @@ 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")):
+        if is_batch_retrieve_call_type(kwargs.get("call_type")) or is_batch_line_item_event(kwargs):
             return
         try:
             """

tests/test_litellm/integrations/clickhouse/test_clickhouse_spend_logger.py

@@ -330,3 +330,62 @@ async def test_trace_ingest_and_invalid_payload_do_not_write_spend():
 
     assert logger.log_queue == []
     storage.ensure_schema.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_batch_line_item_success_event_does_not_write_spend(monkeypatch: pytest.MonkeyPatch):
+    # A batch line item carries call_type=acompletion + litellm_params.batch_parent_id.
+    # The aggregate aretrieve_batch row already bills the batch, so a per-line spend row
+    # would bill it twice.
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    storage = MagicMock()
+    logger = ClickHouseSpendLogger(storage=storage)
+    now = datetime.now(timezone.utc)
+
+    await logger.async_log_success_event(
+        {
+            "standard_logging_object": _minimal_payload("chatcmpl-line-1", status="success", cost=0.25),
+            "call_type": "acompletion",
+            "litellm_params": {"batch_parent_id": "batch-1"},
+        },
+        None,
+        now,
+        now,
+    )
+
+    assert logger.log_queue == []
+
+    await logger.async_log_success_event(
+        {
+            "standard_logging_object": _minimal_payload("chatcmpl-live-1", status="success", cost=0.25),
+            "call_type": "acompletion",
+            "litellm_params": {},
+        },
+        None,
+        now,
+        now,
+    )
+
+    # The aggregate aretrieve_batch event bills the batch; it must still write one
+    # row with the batch's full cost even while line items are skipped.
+    await logger.async_log_success_event(
+        {
+            "standard_logging_object": {
+                **_minimal_payload("batch-1", status="success", cost=0.0001032),
+                "call_type": "aretrieve_batch",
+            },
+            "call_type": "aretrieve_batch",
+            "litellm_params": {},
+        },
+        None,
+        now,
+        now,
+    )
+
+    assert len(logger.log_queue) == 2
+    assert logger.log_queue[0]["request_id"] == "chatcmpl-live-1"
+    assert logger.log_queue[1]["call_type"] == "aretrieve_batch"
+    assert logger.log_queue[1]["request_id"] == "batch-1"
+    assert logger.log_queue[1]["spend"] == 0.0001032
+    if logger._flush_task is not None:
+        logger._flush_task.cancel()

tests/unit/litellm_core_utils/test_litellm_logging.py

@@ -9321,3 +9321,19 @@ class TestRetrieveBatchReusesFetchedResultFiles:
 
         assert batch._hidden_params["response_cost"] == 1.5
         assert batch.usage.total_tokens == 15
+
+
+def test_caller_supplied_litellm_params_cannot_forge_logging_markers(logging_obj):
+    """A request body key `litellm_params` flows into optional_params; the logging
+    kwargs must keep the internal litellm_params or a caller could forge markers
+    (batch_parent_id) that spend sinks and router counters use to skip events."""
+    from litellm.litellm_core_utils.core_helpers import is_batch_line_item_event
+
+    logging_obj.update_environment_variables(
+        litellm_params={"metadata": {"model_group": "gpt-4o"}},
+        optional_params={"litellm_params": {"batch_parent_id": "fake-batch"}},
+    )
+
+    assert logging_obj.model_call_details["litellm_params"] is logging_obj.litellm_params
+    assert "batch_parent_id" not in logging_obj.model_call_details["litellm_params"]
+    assert not is_batch_line_item_event(logging_obj.model_call_details)

tests/unit/router_strategy/test_least_busy.py

@@ -208,3 +208,28 @@ async def test_an_open_circuit_breaker_falls_back_without_a_warning_per_request(
     assert picked is DEPLOYMENT_B
     assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
     assert sum("circuit breaker is open" in record.getMessage() for record in caplog.records) == 2
+
+
+@pytest.mark.asyncio
+async def test_batch_line_item_success_does_not_release_in_flight_slots(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A batch line item is historical batch traffic (call_type=acompletion +
+    litellm_params.batch_parent_id): it never started a live request, so its success
+    callback must not decrement the in-flight count a live request still holds."""
+    import litellm
+
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    shared: Final = SharedRedisCounters()
+    worker: Final = _worker(shared)
+    worker.log_pre_api_call(model="m", messages=[], kwargs=_call_kwargs("dep-a"))
+
+    line_item_kwargs: Final = {
+        "call_type": "acompletion",
+        "litellm_params": {
+            "batch_parent_id": "batch-1",
+            "metadata": {"model_group": GROUP},
+            "model_info": {"id": "dep-a"},
+        },
+    }
+    await worker.async_log_success_event(line_item_kwargs, None, None, None)
+
+    assert shared.count(f"{GROUP}_request_count:dep-a") == 1

tests/unit/router_strategy/test_lowest_tpm_rpm.py

@@ -117,3 +117,40 @@ async def test_v2_subclass_overriding_async_get_available_deployments_with_the_o
         f"from {HIGH_USAGE_DEPLOYMENT_ID}",
         f"from {LOW_USAGE_DEPLOYMENT_ID}",
     }
+
+
+@pytest.mark.asyncio
+async def test_usage_based_routing_handlers_skip_batch_line_items(monkeypatch: pytest.MonkeyPatch) -> None:
+    """
+    Batch line-item callbacks carry call_type=acompletion plus
+    litellm_params.batch_parent_id, so the call-type-only batch guard does not fire.
+    They are historical batch traffic and must not update the TPM/RPM usage counters.
+    """
+    import litellm
+    from litellm.router_strategy.lowest_tpm_rpm import LowestTPMLoggingHandler
+
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    router: Final = Router(
+        model_list=[_deployment(HIGH_USAGE_DEPLOYMENT_ID)],
+        routing_strategy="usage-based-routing",
+    )
+    handler: Final = LowestTPMLoggingHandler(router_cache=router.cache, routing_args={})
+    line_item_kwargs: Final = {
+        "call_type": "acompletion",
+        "litellm_params": {
+            "batch_parent_id": "batch-1",
+            "metadata": {"model_group": MODEL_GROUP},
+            "model_info": {"id": HIGH_USAGE_DEPLOYMENT_ID},
+        },
+    }
+    response_obj: Final = {"usage": {"total_tokens": 600}}
+
+    handler.log_success_event(line_item_kwargs, response_obj, None, None)
+    await handler.async_log_success_event(line_item_kwargs, response_obj, None, None)
+
+    moved: Final = sorted(
+        f"{key}={router.cache.in_memory_cache.cache_dict[key]}"
+        for key in router.cache.in_memory_cache.cache_dict
+        if ":tpm:" in key or ":rpm:" in key
+    )
+    assert moved == []

tests/unit/test_router/test_router.py

@@ -1263,6 +1263,53 @@ def test_sync_deployment_callback_on_success_skips_batch_retrieves(
         == expected_successes
     )
 
+
+@pytest.mark.asyncio
+async def test_deployment_callbacks_skip_batch_line_items(monkeypatch: pytest.MonkeyPatch):
+    """
+    Batch line-item callbacks carry call_type=acompletion plus
+    litellm_params.batch_parent_id, so the call-type-only batch guards do not fire.
+    They describe historical batch traffic already reported by the aggregate
+    aretrieve_batch event and must not consume live TPM/RPM quota.
+    """
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    router = litellm.Router(
+        model_list=[
+            {
+                "model_name": _BATCH_GROUP,
+                "litellm_params": {"model": _BATCH_DEPLOYMENT_MODEL, "api_base": _BATCH_API_BASE, "api_key": "sk-fake"},
+                "model_info": {"id": "batch-dep"},
+            }
+        ]
+    )
+    now = datetime.now()
+    line_item_kwargs = {
+        "call_type": "acompletion",
+        "standard_logging_object": {"total_tokens": _BATCH_TOKENS_PER_ROW},
+        "litellm_params": {
+            "batch_parent_id": _BATCH_ID,
+            "metadata": {"model_group": _BATCH_GROUP, "deployment": _BATCH_DEPLOYMENT_MODEL},
+            "model_info": {"id": "batch-dep"},
+        },
+    }
+
+    await router.deployment_callback_on_success(
+        kwargs=line_item_kwargs, completion_response=None, start_time=now, end_time=now
+    )
+    sync_key = router.sync_deployment_callback_on_success(
+        kwargs=line_item_kwargs, completion_response=None, start_time=now, end_time=now
+    )
+    await router.async_deployment_callback_on_failure(
+        kwargs=line_item_kwargs, completion_response=None, start_time=now, end_time=now
+    )
+
+    assert sync_key is None
+    assert (
+        get_deployment_successes_for_current_minute(litellm_router_instance=router, deployment_id="batch-dep") == 0
+    )
+    assert await _moved_routing_counters(router) == []
+    assert await _router_usage_keys(router) == []
+
 _ROUTING_STRATEGY_CACHE_MARKERS = ("_map", "_request_count", ":tpm:", ":rpm:")
 
 
@@ -1350,6 +1397,81 @@ async def test_arouter_aretrieve_batch_does_not_feed_routing_strategies(
     assert moved_counters == []
 
 
+_BATCH_INPUT_JSONL = "\n".join(
+    json.dumps(
+        {
+            "custom_id": f"row-{row}",
+            "method": "POST",
+            "url": "/v1/chat/completions",
+            "body": {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hi"}]},
+        }
+    )
+    for row in range(_BATCH_ROWS)
+)
+
+
+def _mock_batch_input_file(respx_mock):
+    respx_mock.get(f"{_BATCH_API_BASE}/files/file-in-1/content").mock(
+        return_value=httpx.Response(200, text=_BATCH_INPUT_JSONL)
+    )
+
+
+async def _line_item_payloads(collector: "_BatchPayloadCollector", minimum: int, timeout: float = 5.0) -> list:
+    loop = asyncio.get_event_loop()
+    deadline = loop.time() + timeout
+    while loop.time() < deadline:
+        line_items = [p for p in collector.payloads if p and p.get("call_type") == "acompletion"]
+        if len(line_items) >= minimum:
+            return line_items
+        await asyncio.sleep(0.05)
+    raise AssertionError(f"expected at least {minimum} batch line-item payloads, saw {len(collector.payloads)} total")
+
+
+@pytest.mark.parametrize(
+    "routing_strategy",
+    [
+        "usage-based-routing",
+        "usage-based-routing-v2",
+        "latency-based-routing",
+        "cost-based-routing",
+        "least-busy",
+    ],
+)
+@pytest.mark.asyncio
+async def test_arouter_aretrieve_batch_line_items_do_not_feed_routing_strategies(
+    monkeypatch: pytest.MonkeyPatch, routing_strategy: str
+):
+    """
+    With the line-item flag on, retrieving a completed batch also emits one child
+    callback per JSONL line (call_type=acompletion + litellm_params.batch_parent_id,
+    inheriting the parent deployment's model_info). Those children are historical
+    batch traffic: like the aggregate poll, they must not move the counters that
+    decide where the next live chat request goes.
+    """
+    from litellm.batches.batch_line_item_logging import batch_line_item_claim_cache
+
+    batch_line_item_claim_cache.in_memory_cache.flush_cache()
+    monkeypatch.setattr(litellm, "store_batch_line_items_in_callbacks", True, raising=False)
+    collector = _BatchPayloadCollector()
+    monkeypatch.setattr(litellm, "callbacks", [collector])
+    monkeypatch.setattr(litellm, "input_callback", [])
+    router = _batch_fan_out_router(routing_strategy)
+
+    with respx.mock(assert_all_called=True) as respx_mock:
+        _mock_batch_provider(respx_mock)
+        _mock_batch_input_file(respx_mock)
+        respx_mock.get(f"{_UNRELATED_BATCH_API_BASE}/batches/{_BATCH_ID}").mock(
+            return_value=httpx.Response(404, json=_BATCH_NOT_FOUND)
+        )
+        response = await router.aretrieve_batch(batch_id=_BATCH_ID)
+        line_items = await _line_item_payloads(collector, minimum=_BATCH_ROWS)
+        moved_counters = await _moved_routing_counters(router)
+
+    assert response.id == _BATCH_ID
+    assert len(line_items) == _BATCH_ROWS
+    assert moved_counters == []
+
+
 @pytest.mark.asyncio
 async def test_arouter_aretrieve_file_content():
     """