feat(gemini): plumb through cache tokens in metadata events

Surface cached_content_token_count from usage_metadata as
cacheReadInputTokens on the metadata event emitted by GeminiModel.
The existing telemetry pipeline picks it up automatically.

Relates to #1060, #1140.

Author: yatszhash

src/strands/models/gemini.py

@@ -16,6 +16,7 @@
 from typing_extensions import Required, Unpack, override
 
 from ..types.content import ContentBlock, ContentBlockStartToolUse, Messages, SystemContentBlock
+from ..types.event_loop import Usage
 from ..types.exceptions import ContextWindowOverflowException, ModelThrottledException, ProviderTokenCountError
 from ..types.streaming import StreamEvent
 from ..types.tools import ToolChoice, ToolSpec
@@ -423,13 +424,18 @@ def _format_chunk(self, event: dict[str, Any]) -> StreamEvent:
                         return {"messageStop": {"stopReason": "end_turn"}}
 
             case "metadata":
+                usage_data: Usage = {
+                    "inputTokens": event["data"].prompt_token_count,
+                    "outputTokens": event["data"].total_token_count - event["data"].prompt_token_count,
+                    "totalTokens": event["data"].total_token_count,
+                }
+
+                if cached := event["data"].cached_content_token_count:
+                    usage_data["cacheReadInputTokens"] = cached
+
                 return {
                     "metadata": {
-                        "usage": {
-                            "inputTokens": event["data"].prompt_token_count,
-                            "outputTokens": event["data"].total_token_count - event["data"].prompt_token_count,
-                            "totalTokens": event["data"].total_token_count,
-                        },
+                        "usage": usage_data,
                         "metrics": {
                             "latencyMs": 0,  # TODO
                         },

tests/strands/models/test_gemini.py

@@ -569,6 +569,57 @@ async def test_stream_response_text(gemini_client, model, messages, agenerator,
     assert tru_chunks == exp_chunks
 
 
+def test_format_chunk_metadata_with_cache_tokens(model):
+    """Test _format_chunk for metadata with cache tokens present."""
+    event = {
+        "chunk_type": "metadata",
+        "data": genai.types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            total_token_count=150,
+            cached_content_token_count=25,
+        ),
+    }
+
+    result = model._format_chunk(event)
+
+    assert result == {
+        "metadata": {
+            "usage": {
+                "inputTokens": 100,
+                "outputTokens": 50,
+                "totalTokens": 150,
+                "cacheReadInputTokens": 25,
+            },
+            "metrics": {"latencyMs": 0},
+        },
+    }
+
+
+def test_format_chunk_metadata_with_zero_cached_tokens(model):
+    """Test _format_chunk for metadata when cached_content_token_count is 0."""
+    event = {
+        "chunk_type": "metadata",
+        "data": genai.types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            total_token_count=150,
+            cached_content_token_count=0,
+        ),
+    }
+
+    result = model._format_chunk(event)
+
+    assert result == {
+        "metadata": {
+            "usage": {
+                "inputTokens": 100,
+                "outputTokens": 50,
+                "totalTokens": 150,
+            },
+            "metrics": {"latencyMs": 0},
+        },
+    }
+
+
 @pytest.mark.asyncio
 async def test_stream_response_tool_use(gemini_client, model, messages, agenerator, alist):
     gemini_client.aio.models.generate_content_stream.return_value = agenerator(