From 79a740bec21c8ba8f68a25acac8d3ce2838ddd0c Mon Sep 17 00:00:00 2001 From: Harsh <55448981+Harsh23Kashyap@users.noreply.github.com> Date: Thu, 22 Oct 2026 01:03:54 +0530 Subject: [PATCH 1/2] fix(ai): keep latest cumulative output tokens across message_deltas Server tools such as web search can emit several message_delta events in one Anthropic stream, and each delta's usage is cumulative. The stream accumulator added every delta's output_tokens together, so a stream with two deltas reporting 50 then 110 cumulative tokens captured 160. Keep the latest cumulative value, matching how the input and cache counts are already handled and how the Anthropic SDK accumulates usage. --- .../cumulative-delta-output-tokens.md | 5 ++ posthog/ai/anthropic/_anthropic_stream.py | 10 ++- posthog/test/ai/anthropic/test_anthropic.py | 79 +++++++++++++++++++ 3 files changed, 93 insertions(+), 1 deletion(-) create mode 100644 .sampo/changesets/cumulative-delta-output-tokens.md diff --git a/.sampo/changesets/cumulative-delta-output-tokens.md b/.sampo/changesets/cumulative-delta-output-tokens.md new file mode 100644 index 000000000..cbc8bc720 --- /dev/null +++ b/.sampo/changesets/cumulative-delta-output-tokens.md @@ -0,0 +1,5 @@ +--- +pypi/posthog: patch +--- + +Streamed Anthropic generations no longer stack output tokens when a stream emits more than one `message_delta` (server tools can send several). Each delta's usage is cumulative, so the accumulator now keeps the latest total instead of adding every delta together. diff --git a/posthog/ai/anthropic/_anthropic_stream.py b/posthog/ai/anthropic/_anthropic_stream.py index ff3fbe0b8..2df160ef8 100644 --- a/posthog/ai/anthropic/_anthropic_stream.py +++ b/posthog/ai/anthropic/_anthropic_stream.py @@ -53,7 +53,15 @@ def consume(self, event: Any) -> None: event_usage = extract_anthropic_usage_from_event(event) merge_usage_stats(self.usage_stats, event_usage) if getattr(event, "type", None) == "message_delta": - self._apply_cumulative_input_usage(getattr(event, "usage", None)) + usage = getattr(event, "usage", None) + self._apply_cumulative_input_usage(usage) + output_tokens = getattr(usage, "output_tokens", None) + if isinstance(output_tokens, int) and not isinstance(output_tokens, bool): + # output_tokens on a message_delta is cumulative too, so the + # incremental merge above must not stack successive deltas + # (server tools can emit several in one stream). Keep the + # latest total; a field the event omits leaves the prior one. + self.usage_stats["output_tokens"] = output_tokens if getattr(event, "type", None) == "content_block_start": block, tool = handle_anthropic_content_block_start(event) diff --git a/posthog/test/ai/anthropic/test_anthropic.py b/posthog/test/ai/anthropic/test_anthropic.py index 08c66e7e2..ae1f16cf7 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -907,6 +907,85 @@ async def mock_async_create(**kwargs): assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"]) +def anthropic_stream_with_two_cumulative_deltas(): + # Server tools can emit more than one message_delta in a single stream, + # and each delta's usage is cumulative: later values replace earlier ones + # rather than adding to them. + return [ + RawMessageStartEvent( + type="message_start", + message=Message( + id="msg_1", + type="message", + role="assistant", + model="claude-haiku-4-5", + content=[], + stop_reason=None, + stop_sequence=None, + usage=Usage( + input_tokens=100, + output_tokens=1, + cache_creation_input_tokens=0, + cache_read_input_tokens=0, + ), + ), + ), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="pause_turn", stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=50), + ), + RawMessageDeltaEvent( + type="message_delta", + delta=Delta(stop_reason="end_turn", stop_sequence=None), + usage=MessageDeltaUsage(output_tokens=110), + ), + RawMessageStopEvent(type="message_stop"), + ] + + +def test_streaming_multiple_message_deltas_do_not_stack_output_tokens(mock_client): + with patch( + "anthropic.resources.Messages.create", + return_value=iter(anthropic_stream_with_two_cumulative_deltas()), + ): + client = Anthropic(api_key="test-key", posthog_client=mock_client) + response = client.messages.create( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hello"}], + stream=True, + ) + list(response) + + props = mock_client.capture.call_args.kwargs["properties"] + assert props["$ai_output_tokens"] == 110 + + +@pytest.mark.asyncio +async def test_streaming_multiple_message_deltas_do_not_stack_output_tokens_async(mock_client): + async def mock_async_stream(): + for event in anthropic_stream_with_two_cumulative_deltas(): + yield event + + async def mock_async_create(**kwargs): + return mock_async_stream() + + with patch( + "anthropic.resources.messages.AsyncMessages.create", + side_effect=mock_async_create, + ): + client = AsyncAnthropic(api_key="test-key", posthog_client=mock_client) + response = await client.messages.create( + model="claude-haiku-4-5", + messages=[{"role": "user", "content": "Hello"}], + stream=True, + ) + [event async for event in response] + + props = mock_client.capture.call_args.kwargs["properties"] + assert props["$ai_output_tokens"] == 110 + + def test_anthropic_bedrock_preserves_cache_creation_ttl_breakdown( mock_client, mock_anthropic_response_with_cache_ttl ): From 3844c36634636ddce7033bc464b353773dabe044 Mon Sep 17 00:00:00 2001 From: Manoel Aranda Neto Date: Thu, 22 Oct 2026 12:24:21 +0200 Subject: [PATCH 2/2] style(ai): format cumulative token async regression test --- posthog/test/ai/anthropic/test_anthropic.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/posthog/test/ai/anthropic/test_anthropic.py b/posthog/test/ai/anthropic/test_anthropic.py index ae1f16cf7..ffc6feeed 100644 --- a/posthog/test/ai/anthropic/test_anthropic.py +++ b/posthog/test/ai/anthropic/test_anthropic.py @@ -962,7 +962,9 @@ def test_streaming_multiple_message_deltas_do_not_stack_output_tokens(mock_clien @pytest.mark.asyncio -async def test_streaming_multiple_message_deltas_do_not_stack_output_tokens_async(mock_client): +async def test_streaming_multiple_message_deltas_do_not_stack_output_tokens_async( + mock_client, +): async def mock_async_stream(): for event in anthropic_stream_with_two_cumulative_deltas(): yield event