From c656aa32533fd1432fbff36d666c8a71f2f27e0c Mon Sep 17 00:00:00 2001
From: Yucheng Zhu <yucheng@berri.ai>
Date: Wed, 26 Aug 2026 02:00:21 -0700
Subject: [PATCH] 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 | 10 +++++--
 .../integrations/otel/test_otel_v2_metrics.py | 30 +++++++++++++++++++
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/litellm/integrations/otel/plumbing/metrics.py b/litellm/integrations/otel/plumbing/metrics.py
index 548a6440126..69116ebe20c 100644
--- a/litellm/integrations/otel/plumbing/metrics.py
+++ b/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(
diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_metrics.py b/tests/test_litellm/integrations/otel/test_otel_v2_metrics.py
index b810ffdc6be..016dbcd824b 100644
--- a/tests/test_litellm/integrations/otel/test_otel_v2_metrics.py
+++ b/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."""