test(streaming): narrow authoritative zero usage exclusions

tests/test_litellm/streaming_contract_matrix.md

@@ -70,7 +70,9 @@ Partial usage is asserted for async failure paths where the SDK supplies it. Syn
 | STREAM-004 | Empty string `mock_response` sends a provider request |
 | STREAM-005 | Exact cold native sync/async text OpenAI/Pydantic `MockValSer` serialization error |
 
-Each exclusion is reached only at the affected assertion or exact exception. Other assertions execute first where the stream can complete. STREAM-005 can interrupt the stream, so later assertions in that occurrence remain unexecuted. Its attribution to a particular package is unresolved. No production workaround or schema warming is performed
+Each exclusion is reached only at the affected assertion or exact exception. STREAM-002 accepts only the observed recount for this fixture: a zero prompt count becomes 8, a zero output count becomes 2, and total usage remains their sum. Nonzero counts must remain unchanged. Both caller-visible and callback usage must match either the correct tuple or that exact known tuple before the expected failure is recorded
+
+Other assertions execute first where the stream can complete. STREAM-005 can interrupt the stream, so later assertions in that occurrence remain unexecuted. Its attribution to a particular package is unresolved. No production workaround or schema warming is performed
 
 These cases must not be reported as fully protected. Use the verbose test report or JUnit output to see exact affected combinations. The findings and local mutation evidence are provided separately from the implementation diff
 

tests/test_litellm/test_streaming_contract_matrix.py

@@ -301,14 +301,26 @@ async def test_success(
             field(events[0].response.get("usage"), key)
             for key in ("prompt_tokens", "completion_tokens", "total_tokens")
         )
-        if (prompt_tokens == 0 or output_tokens == 0) and not recounted and callback_usage != expected:
-            pytest.xfail("STREAM-002: provider authoritative zero usage is recounted")
-        assert callback_usage == expected
-        if visible:
-            assert (
-                tuple(field(visible_usage[0], key) for key in ("prompt_tokens", "completion_tokens", "total_tokens"))
-                == expected
+        observed_usages: Final = (callback_usage,) + (
+            tuple(
+                tuple(field(item, key) for key in ("prompt_tokens", "completion_tokens", "total_tokens"))
+                for item in visible_usage
             )
+            if visible
+            else ()
+        )
+        known_recounted: Final = (
+            8 if prompt_tokens == 0 else prompt_tokens,
+            2 if output_tokens == 0 else output_tokens,
+        )
+        allowed_usages: Final = (
+            (expected, (*known_recounted, sum(known_recounted)))
+            if (prompt_tokens == 0 or output_tokens == 0) and not recounted
+            else (expected,)
+        )
+        assert all(usage in allowed_usages for usage in observed_usages), observed_usages
+        if any(usage != expected for usage in observed_usages):
+            pytest.xfail("STREAM-002: provider authoritative zero usage is recounted")
 
 
 @pytest.mark.asyncio