diff --git a/.sampo/changesets/omit-unreported-aux-token-zeros.md b/.sampo/changesets/omit-unreported-aux-token-zeros.md
new file mode 100644
index 000000000..18994e753
--- /dev/null
+++ b/.sampo/changesets/omit-unreported-aux-token-zeros.md
@@ -0,0 +1,5 @@
+---
+pypi/posthog: patch
+---
+
+Streaming generations interrupted before the provider reported any usage no longer send zero `$ai_cache_read_input_tokens`, `$ai_cache_creation_input_tokens`, or `$ai_reasoning_tokens`. A fabricated 0 reads as a report of nothing, so cost processing priced an unknown generation as a known $0.00 instead of leaving it unknown. Streams whose usage was reported keep the historical zero defaults.
diff --git a/posthog/ai/utils.py b/posthog/ai/utils.py
index c456c0b72..895b34ab0 100644
--- a/posthog/ai/utils.py
+++ b/posthog/ai/utils.py
@@ -825,9 +825,14 @@ def capture_streaming_event(
     if available_tools:
         event_properties["$ai_tools"] = available_tools
 
-    # Add optional token fields
+    # Add optional token fields. The zero-defaults below only apply when the
+    # provider reported usage at all: a stream interrupted before any report
+    # has nothing to default, and a fabricated 0 would read as a report of
+    # nothing where absence means unknown.
+    usage_was_reported = input_tokens is not None or output_tokens is not None
+
     # For Anthropic, always include cache fields even if 0 (backward compatibility)
-    if event_data["provider"] == "anthropic":
+    if event_data["provider"] == "anthropic" and usage_was_reported:
         # Anthropic always includes cache fields
         cache_read = event_data["usage_stats"].get("cache_read_input_tokens", 0)
         cache_creation = event_data["usage_stats"].get("cache_creation_input_tokens", 0)
@@ -847,10 +852,15 @@ def capture_streaming_event(
             # OpenAI async streams historically included these fields even when 0.
             # Keep those defaults in the shared path so they are not mistaken for
             # caller-supplied token passthrough properties.
-            if event_data["provider"] == "openai" and field in {
-                "cache_read_input_tokens",
-                "reasoning_tokens",
-            }:
+            if (
+                event_data["provider"] == "openai"
+                and usage_was_reported
+                and field
+                in {
+                    "cache_read_input_tokens",
+                    "reasoning_tokens",
+                }
+            ):
                 event_properties.setdefault(
                     property_name, event_data["usage_stats"].get(field, 0)
                 )
diff --git a/posthog/test/ai/test_token_reporting.py b/posthog/test/ai/test_token_reporting.py
index 165dd42cf..790a3a6db 100644
--- a/posthog/test/ai/test_token_reporting.py
+++ b/posthog/test/ai/test_token_reporting.py
@@ -15,9 +15,11 @@ def mock_client():
         yield mock_client
 
 
-def _event_data(usage_stats: TokenUsage) -> StreamingEventData:
+def _event_data(
+    usage_stats: TokenUsage, provider: str = "gemini"
+) -> StreamingEventData:
     return StreamingEventData(
-        provider="gemini",
+        provider=provider,
         model="gemini-2.0-flash",
         base_url="https://generativelanguage.googleapis.com",
         kwargs={},
@@ -56,3 +58,40 @@ def test_token_counts_trace_back_to_a_provider_report(
             assert props[key] == expected[key]
         else:
             assert key not in props
+
+
+@pytest.mark.parametrize(
+    ("provider", "usage_stats", "expected"),
+    [
+        # A stream interrupted before any usage report has nothing to default:
+        # a fabricated 0 would read as a report of nothing.
+        ("openai", TokenUsage(), {}),
+        ("anthropic", TokenUsage(), {}),
+        # Reported usage keeps the historical zero-defaults.
+        (
+            "openai",
+            TokenUsage(input_tokens=12, output_tokens=7),
+            {"$ai_cache_read_input_tokens": 0, "$ai_reasoning_tokens": 0},
+        ),
+        (
+            "anthropic",
+            TokenUsage(input_tokens=10),
+            {"$ai_cache_read_input_tokens": 0, "$ai_cache_creation_input_tokens": 0},
+        ),
+    ],
+)
+def test_aux_token_fields_trace_back_to_a_provider_report(
+    mock_client, provider, usage_stats, expected
+):
+    capture_streaming_event(mock_client, _event_data(usage_stats, provider=provider))
+
+    props = mock_client.capture.call_args[1]["properties"]
+    for key in (
+        "$ai_cache_read_input_tokens",
+        "$ai_cache_creation_input_tokens",
+        "$ai_reasoning_tokens",
+    ):
+        if key in expected:
+            assert props[key] == expected[key]
+        else:
+            assert key not in props