fix(ai): take cumulative Anthropic stream usage from message_delta

.sampo/changesets/cumulative-anthropic-stream-usage.md

@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Streamed Anthropic generations now take the input and cache token counts from the final `message_delta` usage, which is cumulative. With server tools such as web search, those counts grow past the `message_start` values, so streamed calls under-reported input tokens and cache writes.

posthog/ai/anthropic/_anthropic_stream.py

@@ -26,9 +26,34 @@ def __init__(self) -> None:
         # web_search_tool_result, ...), so keep a list that lines up with it.
         self._blocks_by_index: List[StreamingContentBlock] = []
 
+    def _apply_cumulative_input_usage(self, usage: Any) -> None:
+        # Usage on message_delta is cumulative. With server tools (web search,
+        # web fetch, code execution) its input and cache counts grow past
+        # message_start's, which is all the per-event merge records for them.
+        # The totals never shrink, so keep the larger value; a field the event
+        # omits leaves message_start's count in place.
+        input_tokens = getattr(usage, "input_tokens", None)
+        if isinstance(input_tokens, int):
+            self.usage_stats["input_tokens"] = max(
+                self.usage_stats.get("input_tokens") or 0, input_tokens
+            )
+        cache_creation = getattr(usage, "cache_creation_input_tokens", None)
+        if isinstance(cache_creation, int):
+            self.usage_stats["cache_creation_input_tokens"] = max(
+                self.usage_stats.get("cache_creation_input_tokens") or 0,
+                cache_creation,
+            )
+        cache_read = getattr(usage, "cache_read_input_tokens", None)
+        if isinstance(cache_read, int):
+            self.usage_stats["cache_read_input_tokens"] = max(
+                self.usage_stats.get("cache_read_input_tokens") or 0, cache_read
+            )
+
     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))
 
         if getattr(event, "type", None) == "content_block_start":
             block, tool = handle_anthropic_content_block_start(event)

posthog/test/ai/anthropic/test_anthropic.py

@@ -824,6 +824,89 @@ async def mock_async_create(**kwargs):
     )
 
 
+def anthropic_stream_with_server_tool_usage():
+    # With server tools, message_delta reports cumulative usage that exceeds
+    # message_start's (values from a recorded web search stream).
+    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=2247,
+                    output_tokens=26,
+                    cache_creation_input_tokens=0,
+                    cache_read_input_tokens=0,
+                ),
+            ),
+        ),
+        RawMessageDeltaEvent(
+            type="message_delta",
+            delta=Delta(stop_reason="end_turn", stop_sequence=None),
+            usage=MessageDeltaUsage(
+                input_tokens=2254,
+                output_tokens=110,
+                cache_creation_input_tokens=17269,
+                cache_read_input_tokens=0,
+            ),
+        ),
+        RawMessageStopEvent(type="message_stop"),
+    ]
+
+
+def assert_cumulative_delta_usage(props):
+    assert props["$ai_input_tokens"] == 2254
+    assert props["$ai_output_tokens"] == 110
+    assert props["$ai_cache_creation_input_tokens"] == 17269
+    assert props["$ai_cache_read_input_tokens"] == 0
+
+
+def test_streaming_takes_cumulative_message_delta_usage(mock_client):
+    with patch(
+        "anthropic.resources.Messages.create",
+        return_value=iter(anthropic_stream_with_server_tool_usage()),
+    ):
+        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)
+
+    assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"])
+
+
+@pytest.mark.asyncio
+async def test_streaming_takes_cumulative_message_delta_usage_async(mock_client):
+    async def mock_async_stream():
+        for event in anthropic_stream_with_server_tool_usage():
+            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]
+
+    assert_cumulative_delta_usage(mock_client.capture.call_args.kwargs["properties"])
+
+
 def test_anthropic_bedrock_preserves_cache_creation_ttl_breakdown(
     mock_client, mock_anthropic_response_with_cache_ttl
 ):