From b3722dbeb72e860619e12993001735d41d564df3 Mon Sep 17 00:00:00 2001 From: Arthi Arumugam Date: Sun, 26 Jul 2026 17:06:18 +0530 Subject: [PATCH 1/2] fix(anthropic): stop double-counting output tokens on streamed messages message_delta.usage.output_tokens is the running total for the whole message, and message_start.usage.output_tokens is a partial count already included in it. _process_response_item added the two, so every streamed Anthropic span reported inflated output tokens, inflated total tokens and inflated cost downstream. No exception was raised. The vendor SDK assigns rather than adds (anthropic/lib/streaming/_messages.py:506), as does this repo's own Bedrock streaming wrapper. Replaying the existing test_anthropic_message_streaming_legacy cassette gives 174 output tokens against a ground truth of 171. The existing test asserted input_tokens and the input+output==total identity but never the output value itself, so the bug passed CI. Added that assertion. --- .../instrumentation/anthropic/streaming.py | 13 ++++++------- .../tests/test_messages.py | 1 + 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py index bd2d6dc15d..06a6b37fb1 100644 --- a/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py +++ b/packages/opentelemetry-instrumentation-anthropic/opentelemetry/instrumentation/anthropic/streaming.py @@ -62,13 +62,12 @@ def _process_response_item(item, complete_response): event["finish_reason"] = item.delta.stop_reason if item.usage: if "usage" in complete_response: - item_output_tokens = dict(item.usage).get("output_tokens", 0) - existing_output_tokens = complete_response["usage"].get( - "output_tokens", 0 - ) - complete_response["usage"]["output_tokens"] = ( - item_output_tokens + existing_output_tokens - ) + # message_delta carries the running total for the whole message, + # not an increment, so it replaces the partial count reported by + # message_start rather than adding to it. + item_output_tokens = dict(item.usage).get("output_tokens") + if item_output_tokens is not None: + complete_response["usage"]["output_tokens"] = item_output_tokens else: complete_response["usage"] = dict(item.usage) diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_messages.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_messages.py index e33a24ac6b..a81e95f783 100644 --- a/packages/opentelemetry-instrumentation-anthropic/tests/test_messages.py +++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_messages.py @@ -680,6 +680,7 @@ def test_anthropic_message_streaming_legacy( assert text_parts[0]["content"] == response_content assert output_messages[-1]["role"] == "assistant" assert anthropic_span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] == 17 + assert anthropic_span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] == 171 assert ( anthropic_span.attributes[GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS] + anthropic_span.attributes[GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS] From 36a043a890748d9e634589cce2fb3911e1a928bc Mon Sep 17 00:00:00 2001 From: Arthi Arumugam <62213284+arthi-arumugam-git@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:07:12 +0530 Subject: [PATCH 2/2] test(anthropic): drop one-token overcount in prompt-caching stream asserts --- .../tests/test_prompt_caching.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/packages/opentelemetry-instrumentation-anthropic/tests/test_prompt_caching.py b/packages/opentelemetry-instrumentation-anthropic/tests/test_prompt_caching.py index 82a0254383..4e94d113af 100644 --- a/packages/opentelemetry-instrumentation-anthropic/tests/test_prompt_caching.py +++ b/packages/opentelemetry-instrumentation-anthropic/tests/test_prompt_caching.py @@ -759,7 +759,7 @@ def test_anthropic_prompt_caching_stream_legacy( assert json.loads(cache_creation_span.attributes[GenAIAttributes.GEN_AI_OUTPUT_MESSAGES])[-1]["role"] == "assistant" assert json.loads(cache_read_span.attributes[GenAIAttributes.GEN_AI_OUTPUT_MESSAGES])[-1]["role"] == "assistant" - _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 202, 222, 1165) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 201, 221, 1165) # verify metrics metrics_data = reader.get_metrics_data() @@ -820,7 +820,7 @@ def test_anthropic_prompt_caching_stream_with_events_with_content( cache_creation_span = spans[0] cache_read_span = spans[1] - _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 202, 222, 1165) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 201, 221, 1165) # verify metrics metrics_data = reader.get_metrics_data() @@ -986,7 +986,7 @@ def test_anthropic_prompt_caching_stream_with_events_with_no_content( cache_creation_span = spans[0] cache_read_span = spans[1] - _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 202, 222, 1165) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1169, 201, 221, 1165) # verify metrics metrics_data = reader.get_metrics_data() @@ -1089,7 +1089,7 @@ async def test_anthropic_prompt_caching_async_stream_legacy( assert cr_input[0]["role"] == "user" assert text == cr_input[0]["parts"][0]["content"] - _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 290, 257, 1167) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 289, 256, 1167) # verify metrics metrics_data = reader.get_metrics_data() @@ -1151,7 +1151,7 @@ async def test_anthropic_prompt_caching_async_stream_with_events_with_content( cache_creation_span = spans[0] cache_read_span = spans[1] - _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 290, 257, 1167) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 289, 256, 1167) # verify metrics metrics_data = reader.get_metrics_data() @@ -1328,7 +1328,7 @@ async def test_anthropic_prompt_caching_async_stream_with_events_with_no_content cache_creation_span = spans[0] cache_read_span = spans[1] - _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 290, 257, 1167) + _verify_caching_attributes(cache_creation_span, cache_read_span, 1171, 289, 256, 1167) # verify metrics metrics_data = reader.get_metrics_data() @@ -1372,3 +1372,4 @@ def assert_message_in_logs(log: ReadableLogRecord, event_name: str, expected_con else: assert log.log_record.body assert dict(log.log_record.body) == expected_content +