fix: honor server-reported Gemini cache expiry

The client derived cache expiry from the local clock right after creation, but network delay or server-side normalization can push the real expiry off that estimate and lead to reuse at the wrong boundary. This records the expire_time the API returns when present, and falls back to the local TTL only when the response omits it.

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 947961020

Author: GWeale

src/google/adk/models/gemini_context_cache_manager.py

@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+from datetime import datetime
 import hashlib
 import json
 import logging
@@ -532,14 +533,20 @@ async def _create_gemini_cache(
       )
       # Set precise creation timestamp right after cache creation
       created_at = time.time()
+      server_expire_time = getattr(cached_content, "expire_time", None)
+      expire_time = (
+          server_expire_time.timestamp()
+          if isinstance(server_expire_time, datetime)
+          else created_at + llm_request.cache_config.ttl_seconds
+      )
       logger.info("Cache created successfully: %s", cached_content.name)
 
       span.set_attribute("cache_name", cached_content.name)
 
       # Return complete cache metadata with precise timing
       return CacheMetadata(
           cache_name=cached_content.name,
-          expire_time=created_at + llm_request.cache_config.ttl_seconds,
+          expire_time=expire_time,
           fingerprint=self._generate_cache_fingerprint(
               llm_request, cache_contents_count
           ),

tests/unittests/agents/test_gemini_context_cache_manager.py

@@ -14,6 +14,8 @@
 
 """Tests for GeminiContextCacheManager."""
 
+from datetime import datetime
+from datetime import timezone
 import time
 from unittest.mock import AsyncMock
 from unittest.mock import MagicMock
@@ -1157,6 +1159,25 @@ async def test_dynamic_instruction_does_not_break_initial_cache_fingerprint(
     assert result_2.invocations_used == 1
     self.manager.genai_client.aio.caches.create.assert_called_once()
 
+  async def test_create_cache_uses_server_expire_time(self):
+    """The server-reported expiry is authoritative when it is available."""
+    server_expire_time = datetime.fromtimestamp(2_000_000_000, tz=timezone.utc)
+    mock_cached_content = types.CachedContent(
+        name="projects/test/locations/us-central1/cachedContents/test123",
+        expire_time=server_expire_time,
+    )
+    self.manager.genai_client.aio.caches.create = AsyncMock(
+        return_value=mock_cached_content
+    )
+    llm_request = self.create_llm_request()
+
+    with patch.object(
+        self.manager, "_generate_cache_fingerprint", return_value="test_fp"
+    ):
+      cache_metadata = await self.manager._create_gemini_cache(llm_request, 2)
+
+    assert cache_metadata.expire_time == server_expire_time.timestamp()
+
   async def test_create_http_options_passthrough(self):
     """Test that create_http_options is passed through to cache creation config."""
     mock_cached_content = AsyncMock()