fix: grow eligible Gemini cache prefixes

When a cache expired, renewal recreated the original prefix, so completed turns could stay outside every refreshed cache. It also assumed a 4,096-token floor for every model, which skipped eligible Gemini 2.5 prefixes between 2,048 and 4,095 tokens. This grows the cache to the latest validated prefix and applies the documented 2,048-token floor for Gemini 2.5 while keeping 4,096 for Gemini 3.

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

Author: GWeale

src/google/adk/agents/context_cache_config.py

@@ -36,8 +36,9 @@ class ContextCacheConfig(BaseModel):
   by reusing previously processed context across multiple requests.
 
   Caching begins on the second turn of a session at the earliest and requires
-  the prior request to reach Gemini's hard 4096-token minimum, so short or
-  single-turn sessions are never cached.
+  the cacheable prefix to reach the model-specific minimum: 2048 tokens for
+  Gemini 2.5 or 4096 tokens for Gemini 3. Short or single-turn sessions are
+  therefore never cached.
 
   Attributes:
       cache_intervals: Maximum number of invocations to reuse the same cache before refreshing it
@@ -71,11 +72,11 @@ class ContextCacheConfig(BaseModel):
       description=(
           "Minimum prior-request tokens required to enable caching. This gates"
           " on the previous request's actual prompt token count, not an"
-          " estimate of the current request. Gemini enforces a hard 4096-token"
-          " minimum that always applies, so values below 4096 have no"
-          " additional effect. No cache is created on the first request of a"
-          " session; caching begins on the second turn once a previous token"
-          " count is known. Set higher to avoid caching small requests where"
+          " estimate of the current request. Gemini's model-specific minimum"
+          " always applies: 2048 tokens for Gemini 2.5 and 4096 tokens for"
+          " Gemini 3. No cache is created on the first request of a session;"
+          " caching begins on the second turn once a previous token count is"
+          " known. Set this higher to avoid caching small requests where"
           " storage overhead may exceed benefits."
       ),
   )

src/google/adk/models/gemini_context_cache_manager.py

@@ -34,13 +34,25 @@
 
 logger = logging.getLogger("google_adk." + __name__)
 
-# Gemini API requires a minimum of 4096 tokens for cached content.
-_GEMINI_MIN_CACHE_TOKENS = 4096
+# Named Gemini model families have documented explicit-cache floors. For
+# opaque tuned-model and endpoint IDs, the server remains authoritative.
+_GEMINI_2_5_MIN_CACHE_TOKENS = 2048
+_GEMINI_3_MIN_CACHE_TOKENS = 4096
 
 if TYPE_CHECKING:
   from google.genai import Client
 
 
+def _minimum_cache_tokens(model: Optional[str]) -> Optional[int]:
+  """Return the explicit-cache token floor for a named Gemini model."""
+  model_name = (model or "").rsplit("/", maxsplit=1)[-1]
+  if model_name.startswith("gemini-2.5-"):
+    return _GEMINI_2_5_MIN_CACHE_TOKENS
+  if model_name.startswith("gemini-3"):
+    return _GEMINI_3_MIN_CACHE_TOKENS
+  return None
+
+
 @experimental
 class GeminiContextCacheManager:
   """Manages context cache lifecycle for Gemini models.
@@ -104,17 +116,27 @@ async def handle_context_caching(
           )
           await self.cleanup_cache(old_cache_metadata.cache_name)
 
-        # Calculate current fingerprint using contents count from old metadata
-        cache_contents_count = old_cache_metadata.contents_count
+        # Validate the previously fingerprinted prefix before growing it.
+        previous_cache_contents_count = old_cache_metadata.contents_count
         current_fingerprint = self._generate_cache_fingerprint(
-            llm_request, cache_contents_count
+            llm_request, previous_cache_contents_count
         )
 
         # If fingerprints match, create new cache (expired but same content)
         if current_fingerprint == old_cache_metadata.fingerprint:
           logger.debug(
               "Fingerprints match after invalidation, creating new cache"
           )
+          current_cacheable_contents_count = (
+              self._find_count_of_contents_to_cache(llm_request.contents)
+          )
+          cache_contents_count = max(
+              previous_cache_contents_count,
+              current_cacheable_contents_count,
+          )
+          current_fingerprint = self._generate_cache_fingerprint(
+              llm_request, cache_contents_count
+          )
           cache_metadata = await self._create_new_cache_with_contents(
               llm_request, cache_contents_count
           )
@@ -124,9 +146,8 @@ async def handle_context_caching(
             )
             return cache_metadata
 
-          # Cache creation failed (e.g., below Gemini's 4096 token minimum).
-          # Preserve the original contents_count so the fingerprint stays
-          # stable for subsequent calls instead of resetting to total.
+          # Cache creation failed (for example, below the model's minimum).
+          # Preserve the largest stable prefix for the next attempt.
           logger.debug(
               "Cache creation failed, preserving prefix fingerprint "
               "(contents_count=%d)",
@@ -358,11 +379,15 @@ async def _create_new_cache_with_contents(
     cacheable_prefix_tokens = self._estimate_cacheable_prefix_tokens(
         llm_request, cache_contents_count
     )
-    if cacheable_prefix_tokens < _GEMINI_MIN_CACHE_TOKENS:
+    minimum_cache_tokens = _minimum_cache_tokens(llm_request.model)
+    if (
+        minimum_cache_tokens is not None
+        and cacheable_prefix_tokens < minimum_cache_tokens
+    ):
       logger.info(
           "Cacheable prefix below Gemini minimum cache size (%d < %d tokens)",
           cacheable_prefix_tokens,
-          _GEMINI_MIN_CACHE_TOKENS,
+          minimum_cache_tokens,
       )
       return None
 

tests/unittests/agents/test_gemini_context_cache_manager.py

@@ -304,6 +304,109 @@ async def test_create_cache_gates_on_prefix_not_full_prompt(self):
     assert result is None
     self.manager.genai_client.aio.caches.create.assert_not_called()
 
+  async def test_completed_turn_grows_cacheable_prefix(self):
+    """A completed turn becomes part of the next explicit cache."""
+    first_user = types.Content(
+        role="user", parts=[types.Part(text="First question")]
+    )
+    first_model = types.Content(
+        role="model", parts=[types.Part(text="First answer")]
+    )
+    next_user = types.Content(
+        role="user", parts=[types.Part(text="Next question")]
+    )
+    first_request = self.create_llm_request(contents_count=0)
+    first_request.contents = [first_user]
+
+    first_metadata = await self.manager.handle_context_caching(first_request)
+
+    assert first_metadata is not None
+    assert first_metadata.contents_count == 0
+
+    next_request = self.create_llm_request(
+        cache_metadata=first_metadata, contents_count=0
+    )
+    next_request.contents = [first_user, first_model, next_user]
+    next_request.cacheable_contents_token_count = 30_000
+    cached_content = AsyncMock()
+    cached_content.name = "cachedContents/grown-prefix"
+    self.manager.genai_client.aio.caches.create = AsyncMock(
+        return_value=cached_content
+    )
+
+    next_metadata = await self.manager.handle_context_caching(next_request)
+
+    assert next_metadata is not None
+    assert next_metadata.cache_name == "cachedContents/grown-prefix"
+    assert next_metadata.contents_count == 2
+    create_config = (
+        self.manager.genai_client.aio.caches.create.call_args.kwargs["config"]
+    )
+    assert create_config.contents == [first_user, first_model]
+    assert next_request.contents == [next_user]
+
+  async def test_gemini_25_creates_cache_above_2048_token_minimum(self):
+    """Gemini 2.5 creates an explicit cache above its 2,048-token floor."""
+    llm_request = self.create_llm_request(contents_count=0)
+    llm_request.config.system_instruction = "x" * 12_000
+    llm_request.cacheable_contents_token_count = 3_000
+    llm_request.cache_metadata = CacheMetadata(
+        fingerprint=self.manager._generate_cache_fingerprint(llm_request, 0),
+        contents_count=0,
+    )
+    cached_content = AsyncMock()
+    cached_content.name = "cachedContents/gemini-25"
+    self.manager.genai_client.aio.caches.create = AsyncMock(
+        return_value=cached_content
+    )
+
+    result = await self.manager.handle_context_caching(llm_request)
+
+    assert result is not None
+    assert result.cache_name == "cachedContents/gemini-25"
+    self.manager.genai_client.aio.caches.create.assert_awaited_once()
+
+  async def test_gemini_3_skips_cache_below_4096_token_minimum(self):
+    """Gemini 3 skips an explicit cache below its 4,096-token floor."""
+    llm_request = self.create_llm_request(contents_count=0)
+    llm_request.model = "gemini-3.1-pro-preview"
+    llm_request.config.system_instruction = "x" * 12_000
+    llm_request.cacheable_contents_token_count = 3_000
+    llm_request.cache_metadata = CacheMetadata(
+        fingerprint=self.manager._generate_cache_fingerprint(llm_request, 0),
+        contents_count=0,
+    )
+
+    result = await self.manager.handle_context_caching(llm_request)
+
+    assert result is not None
+    assert result.cache_name is None
+    self.manager.genai_client.aio.caches.create.assert_not_called()
+
+  async def test_opaque_model_does_not_apply_guessed_token_minimum(self):
+    """Opaque tuned-model IDs let the server enforce the cache floor."""
+    llm_request = self.create_llm_request(contents_count=0)
+    llm_request.model = (
+        "projects/test/locations/us-central1/endpoints/tuned-model"
+    )
+    llm_request.config.system_instruction = "x" * 12_000
+    llm_request.cacheable_contents_token_count = 3_000
+    llm_request.cache_metadata = CacheMetadata(
+        fingerprint=self.manager._generate_cache_fingerprint(llm_request, 0),
+        contents_count=0,
+    )
+    cached_content = AsyncMock()
+    cached_content.name = "cachedContents/tuned-model"
+    self.manager.genai_client.aio.caches.create = AsyncMock(
+        return_value=cached_content
+    )
+
+    result = await self.manager.handle_context_caching(llm_request)
+
+    assert result is not None
+    assert result.cache_name == "cachedContents/tuned-model"
+    self.manager.genai_client.aio.caches.create.assert_awaited_once()
+
   async def test_handle_context_caching_invalid_cache_fingerprint_mismatch(
       self,
   ):
@@ -1155,9 +1258,12 @@ async def test_dynamic_instruction_does_not_break_initial_cache_fingerprint(
     assert result_2.cache_name == (
         "projects/test/locations/us-central1/cachedContents/new789"
     )
-    assert result_2.contents_count == 0
+    assert result_2.contents_count == 2
     assert result_2.invocations_used == 1
-    self.manager.genai_client.aio.caches.create.assert_called_once()
+    create_config = (
+        self.manager.genai_client.aio.caches.create.call_args.kwargs["config"]
+    )
+    assert create_config.contents == [user_msg, model_tool_call]
 
   async def test_create_cache_uses_server_expire_time(self):
     """The server-reported expiry is authoritative when it is available."""