fix(streaming): preserve unknown usage instead of fabricating input_tokens=0

When both message_start and message_delta.input_tokens omit the input
count, the accumulator reported input_tokens=0, which under-reports
accounting to callers. Per review feedback, leave the count unset when
it is genuinely unknown: construct_type already leaves fields the delta
did not supply as None, matching how the rest of the SDK represents
wire-omitted values. A later delta that does supply the count still
fills it in via the accumulate branch.

Author: chenlichao

src/anthropic/lib/streaming/_beta_messages.py

@@ -563,11 +563,11 @@ def accumulate_event(
             # `message_start` may omit usage (see the streaming docs), in which
             # case the snapshot has no usage yet. Initialize it from the delta
             # so the final message still carries token counts, and tolerate
-            # streams that never supply usage.
-            usage = event.usage.to_dict()
-            if event.usage.input_tokens is None:
-                usage["input_tokens"] = 0
-            current_snapshot.usage = construct_type(type_=BetaUsage, value=usage)
+            # streams that never supply usage. Anything the delta leaves out
+            # (e.g. `input_tokens`) stays unset instead of being fabricated
+            # as 0 so an unknown count is never reported to callers; a later
+            # delta that does supply it fills it in below.
+            current_snapshot.usage = construct_type(type_=BetaUsage, value=event.usage.to_dict())
         else:
             current_snapshot.usage.output_tokens = event.usage.output_tokens
 

src/anthropic/lib/streaming/_messages.py

@@ -523,11 +523,11 @@ def accumulate_event(
             # `message_start` may omit usage (see the streaming docs), in which
             # case the snapshot has no usage yet. Initialize it from the delta
             # so the final message still carries token counts, and tolerate
-            # streams that never supply usage.
-            usage = event.usage.to_dict()
-            if event.usage.input_tokens is None:
-                usage["input_tokens"] = 0
-            current_snapshot.usage = construct_type(type_=Usage, value=usage)
+            # streams that never supply usage. Anything the delta leaves out
+            # (e.g. `input_tokens`) stays unset instead of being fabricated
+            # as 0 so an unknown count is never reported to callers; a later
+            # delta that does supply it fills it in below.
+            current_snapshot.usage = construct_type(type_=Usage, value=event.usage.to_dict())
         else:
             current_snapshot.usage.output_tokens = event.usage.output_tokens
 

tests/lib/streaming/fixtures/beta_usage_omitted_input_tokens_response.txt

@@ -0,0 +1,17 @@
+event: message_start
+data: {"type":"message_start","message":{"id":"msg_usage_omitted_input","type":"message","role":"assistant","content":[],"model":"claude-test","stop_reason":null,"stop_sequence":null}}
+
+event: content_block_start
+data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":6}}
+
+event: message_stop
+data: {"type":"message_stop"}

tests/lib/streaming/fixtures/usage_omitted_input_tokens_response.txt

@@ -0,0 +1,17 @@
+event: message_start
+data: {"type":"message_start","message":{"id":"msg_usage_omitted_input","type":"message","role":"assistant","content":[],"model":"claude-test","stop_reason":null,"stop_sequence":null}}
+
+event: content_block_start
+data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
+
+event: content_block_delta
+data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
+
+event: content_block_stop
+data: {"type":"content_block_stop","index":0}
+
+event: message_delta
+data: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":6}}
+
+event: message_stop
+data: {"type":"message_stop"}

tests/lib/streaming/test_beta_messages.py

@@ -590,6 +590,29 @@ def test_usage_omitted_at_message_start_uses_beta_usage_and_preserves_delta_opti
         assert message.usage.fallback_credit is not None
         assert message.usage.fallback_credit.status.type == "redeemed"
 
+    @pytest.mark.respx(base_url=base_url)
+    def test_usage_omitted_at_message_start_preserves_unknown_input_tokens(self, respx_mock: MockRouter) -> None:
+        # If both `message_start` and `message_delta` omit the input count then
+        # it is genuinely unknown; the accumulator must preserve that rather
+        # than fabricate `input_tokens=0`, which would under-report accounting.
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(200, content=get_response("beta_usage_omitted_input_tokens_response.txt"))
+        )
+
+        client = Anthropic(base_url=base_url, api_key=api_key)
+
+        with client.beta.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-test",
+        ) as stream:
+            message = stream.get_final_message()
+
+        assert isinstance(message.usage, BetaUsage)
+        # unknown counts stay unset instead of being reported as 0
+        assert message.usage.input_tokens is None  # pyright: ignore[reportUnnecessaryComparison]
+        assert message.usage.output_tokens == 6
+
     @pytest.mark.respx(base_url=base_url)
     @pytest.mark.parametrize("fixture, expected", INPUT_TRANSFORMATIONS_CASES)
     def test_input_transformations_propagated(

tests/lib/streaming/test_messages.py

@@ -411,6 +411,31 @@ def test_usage_omitted_at_message_start(self, respx_mock: MockRouter) -> None:
         assert message.usage.input_tokens == 12
         assert message.usage.output_tokens == 6
 
+    @pytest.mark.respx(base_url=base_url)
+    def test_usage_omitted_at_message_start_preserves_unknown_input_tokens(self, respx_mock: MockRouter) -> None:
+        # If both `message_start` and `message_delta` omit the input count then
+        # it is genuinely unknown; the accumulator must preserve that rather
+        # than fabricate `input_tokens=0`, which would under-report accounting.
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(200, content=get_response("usage_omitted_input_tokens_response.txt"))
+        )
+
+        # A default (non-strict) client mirrors how the docs' event sequence
+        # reaches the accumulator without response-validation rejecting it.
+        client = Anthropic(base_url=base_url, api_key=api_key)
+
+        with client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-test",
+        ) as stream:
+            message = stream.get_final_message()
+
+        assert message.usage is not None
+        # unknown counts stay unset instead of being reported as 0
+        assert message.usage.input_tokens is None  # pyright: ignore[reportUnnecessaryComparison]
+        assert message.usage.output_tokens == 6
+
     @pytest.mark.respx(base_url=base_url)
     def test_usage_omitted_at_message_start_preserves_delta_optional_usage_fields(
         self, respx_mock: MockRouter
@@ -662,6 +687,34 @@ async def test_usage_omitted_at_message_start(self, respx_mock: MockRouter) -> N
         assert message.usage.input_tokens == 12
         assert message.usage.output_tokens == 6
 
+    @pytest.mark.asyncio
+    @pytest.mark.respx(base_url=base_url)
+    async def test_usage_omitted_at_message_start_preserves_unknown_input_tokens(self, respx_mock: MockRouter) -> None:
+        # If both `message_start` and `message_delta` omit the input count then
+        # it is genuinely unknown; the accumulator must preserve that rather
+        # than fabricate `input_tokens=0`, which would under-report accounting.
+        respx_mock.post("/v1/messages").mock(
+            return_value=httpx2.Response(
+                200, content=to_async_iter(get_response("usage_omitted_input_tokens_response.txt"))
+            )
+        )
+
+        # A default (non-strict) client mirrors how the docs' event sequence
+        # reaches the accumulator without response-validation rejecting it.
+        client = AsyncAnthropic(base_url=base_url, api_key=api_key)
+
+        async with client.messages.stream(
+            max_tokens=1024,
+            messages=[{"role": "user", "content": "Say hello there!"}],
+            model="claude-test",
+        ) as stream:
+            message = await stream.get_final_message()
+
+        assert message.usage is not None
+        # unknown counts stay unset instead of being reported as 0
+        assert message.usage.input_tokens is None  # pyright: ignore[reportUnnecessaryComparison]
+        assert message.usage.output_tokens == 6
+
 
 def test_message_delta_fields_are_all_accumulated() -> None:
     # tripwire: handle a new field in accumulate_event (src/anthropic/lib/streaming/_messages.py), then list it here