fix(otel): stop the v2 metrics recorder reporting replayed tokens on response reads

The v2 span builder sources usage from the standard logging payload, so the
earlier fix already zeroes it there. The metrics recorder reads response_obj
directly, so a responses-management read still recorded the original
generation's tokens into gen_ai.client.token.usage and divided generation time
by them for gen_ai.server.time_per_output_token.

The read still records operation and response duration, under the
litellm.responses_management operation, so it stays observable.

litellm/integrations/otel/plumbing/metrics.py

@@ -21,6 +21,7 @@ from litellm.integrations.opentelemetry import (
     METRIC_METADATA_KEYS,
     TOKEN_TYPE_ATTRIBUTE,
     _build_metric_attribute_filter,
+    _is_unbilled_non_inference,
     _resolve_metric_attribute_filter,
 )
 from litellm.integrations.otel.model.metadata import time_to_first_chunk_seconds
@@ -198,16 +199,21 @@ class GenAIMetricRecorder:
     ) -> None:
         common_attrs: Final = self._filter_attributes(self._bounded_attributes(kwargs))
         duration_s: Final = (end_time - start_time).total_seconds()
+        usage_is_replayed: Final = _is_unbilled_non_inference(
+            kwargs.get("call_type"), kwargs.get("litellm_params"), response_obj
+        )
 
         self._metrics.operation_duration.record(duration_s, attributes=common_attrs)
-        self._record_token_usage(response_obj, common_attrs)
+        if not usage_is_replayed:
+            self._record_token_usage(response_obj, common_attrs)
 
         cost: Final = kwargs.get("response_cost")
         if cost:
             self._metrics.token_cost.record(cost, attributes=common_attrs)
 
         self._record_time_to_first_token(kwargs, common_attrs)
-        self._record_time_per_output_token(kwargs, response_obj, end_time, duration_s, common_attrs)
+        if not usage_is_replayed:
+            self._record_time_per_output_token(kwargs, response_obj, end_time, duration_s, common_attrs)
         self._record_response_duration(kwargs, end_time, common_attrs)
 
     def record_failure(

tests/test_litellm/integrations/otel/test_otel_v2_metrics.py

@@ -201,6 +201,36 @@ def test_time_to_first_token_is_streaming_only():
     assert names == set(ALL_METRICS) - {TIME_TO_FIRST_TOKEN}
 
 
+def test_response_read_does_not_replay_the_generation_usage():
+    """A responses-management read returns the ORIGINAL generation's usage on the
+    object it fetches. Recording it would add those tokens again on every poll, so
+    the two usage-derived instruments are skipped while the duration ones, which
+    describe the read itself, still fire."""
+    metrics = _drive_success(InMemoryMetricReader(), call_type="aget_responses")
+
+    assert TOKEN_USAGE not in metrics
+    assert TIME_PER_OUTPUT_TOKEN not in metrics
+    assert OPERATION_DURATION in metrics
+    assert RESPONSE_DURATION in metrics
+
+
+def test_background_response_read_still_records_usage():
+    """A background=true create returns no usage, so its completed read is the only
+    place the generation's tokens are ever seen. Skipping it would lose them
+    entirely rather than deduplicate them."""
+    reader = InMemoryMetricReader()
+    logger = _logger(reader, enable_metrics=True)
+    kwargs, response_obj, start, end = _build_call(call_type="aget_responses")
+    response_obj["background"] = True
+    asyncio.run(logger.async_log_success_event(kwargs, response_obj, start, end))
+
+    metrics = _metrics_by_name(reader)
+    by_type = {dp.attributes[TOKEN_TYPE]: dp for dp in metrics[TOKEN_USAGE]}
+    assert by_type["input"].sum == PROMPT_TOKENS
+    assert by_type["output"].sum == COMPLETION_TOKENS
+    assert TIME_PER_OUTPUT_TOKEN in metrics
+
+
 def test_metrics_disabled_records_nothing():
     """enable_metrics=False: the recorder is never built, so the injected reader
     sees no gen_ai.client.* series even though the success hook runs."""