fix(models): gate Gemini cache creation on cacheable prefix tokens

`_create_new_cache_with_contents` gated on `cacheable_contents_token_count`
(the previous prompt's full `prompt_token_count`) but `caches.create` only
receives the prefix `contents[:cache_contents_count]`. On long conversations
the full count clears Gemini's 4096-token minimum while the cached prefix is
below it, so `caches.create` fails with 400 INVALID_ARGUMENT.

Gate on the prefix that is actually cached: `_estimate_cacheable_prefix_tokens`
scales the accurate full-prompt count by the prefix's estimated share of the
request (reusing `_estimate_request_tokens`, now accepting an optional
`cache_contents_count`). When the prefix spans the whole request the factor is
1.0 and behavior is unchanged.

Fixes #5847

Author: nyxst4ck

src/google/adk/models/gemini_context_cache_manager.py

@@ -320,11 +320,20 @@ async def _create_new_cache_with_contents(
       )
       return None
 
-    # Check client-side to avoid unnecessary API round-trips.
-    if llm_request.cacheable_contents_token_count < _GEMINI_MIN_CACHE_TOKENS:
+    # `cacheable_contents_token_count` is the token count of the whole previous
+    # prompt (system instruction + tools + every content). The cache, however,
+    # only stores the prefix `contents[:cache_contents_count]` plus the system
+    # instruction and tools (see `_create_gemini_cache`). On a long conversation
+    # the full-prompt count can clear Gemini's minimum while the cached prefix is
+    # far smaller, which makes `caches.create` fail with 400 INVALID_ARGUMENT.
+    # Gate on the estimated prefix size so we never send a sub-minimum payload.
+    cacheable_prefix_tokens = self._estimate_cacheable_prefix_tokens(
+        llm_request, cache_contents_count
+    )
+    if cacheable_prefix_tokens < _GEMINI_MIN_CACHE_TOKENS:
       logger.info(
-          "Request below Gemini minimum cache size (%d < %d tokens)",
-          llm_request.cacheable_contents_token_count,
+          "Cacheable prefix below Gemini minimum cache size (%d < %d tokens)",
+          cacheable_prefix_tokens,
           _GEMINI_MIN_CACHE_TOKENS,
       )
       return None
@@ -336,13 +345,20 @@ async def _create_new_cache_with_contents(
       logger.warning("Failed to create cache: %s", e)
       return None
 
-  def _estimate_request_tokens(self, llm_request: LlmRequest) -> int:
-    """Estimate token count for the request.
+  def _estimate_request_tokens(
+      self,
+      llm_request: LlmRequest,
+      cache_contents_count: Optional[int] = None,
+  ) -> int:
+    """Estimate token count for the request (or its cacheable prefix).
 
     This is a rough estimation based on content text length.
 
     Args:
         llm_request: Request to estimate tokens for
+        cache_contents_count: When provided, only the first
+            ``cache_contents_count`` contents are counted (the prefix that gets
+            cached); the system instruction and tools are always included.
 
     Returns:
         Estimated token count
@@ -360,15 +376,54 @@ def _estimate_request_tokens(self, llm_request: LlmRequest) -> int:
           tool_str = json.dumps(tool.model_dump())
           total_chars += len(tool_str)
 
-    # Contents
-    for content in llm_request.contents:
+    # Contents (optionally limited to the cacheable prefix)
+    contents = llm_request.contents
+    if cache_contents_count is not None:
+      contents = contents[:cache_contents_count]
+    for content in contents:
       for part in content.parts:
         if part.text:
           total_chars += len(part.text)
 
     # Rough estimate: 4 characters per token
     return total_chars // 4
 
+  def _estimate_cacheable_prefix_tokens(
+      self, llm_request: LlmRequest, cache_contents_count: int
+  ) -> int:
+    """Estimate the token count of the prefix that will actually be cached.
+
+    The only accurate token count available is
+    ``cacheable_contents_token_count``, which covers the entire previous prompt.
+    Since the cache stores just the prefix ``contents[:cache_contents_count]``
+    (plus system instruction and tools), we scale that accurate count by the
+    prefix's estimated share of the request. When the prefix already spans the
+    whole request the scale factor is 1 and the accurate count is returned
+    unchanged.
+
+    Args:
+        llm_request: Request to estimate the cacheable prefix tokens for
+        cache_contents_count: Number of leading contents that get cached
+
+    Returns:
+        Estimated token count of the cacheable prefix
+    """
+    full_tokens = llm_request.cacheable_contents_token_count
+    if not full_tokens:
+      return 0
+
+    full_estimate = self._estimate_request_tokens(llm_request)
+    if full_estimate <= 0:
+      # No text to estimate from (e.g. non-text parts); fall back to the
+      # accurate full count rather than incorrectly skipping the cache.
+      return full_tokens
+
+    prefix_estimate = self._estimate_request_tokens(
+        llm_request, cache_contents_count
+    )
+    ratio = min(1.0, prefix_estimate / full_estimate)
+    return int(full_tokens * ratio)
+
   async def _create_gemini_cache(
       self, llm_request: LlmRequest, cache_contents_count: int
   ) -> CacheMetadata:

tests/unittests/agents/test_gemini_context_cache_manager.py

@@ -41,15 +41,22 @@ def setup_method(self):
         min_tokens=0,  # Allow caching for tests
     )
 
-  def create_llm_request(self, cache_metadata=None, contents_count=3):
-    """Helper to create test LlmRequest."""
+  def create_llm_request(
+      self, cache_metadata=None, contents_count=3, message_text=None
+  ):
+    """Helper to create test LlmRequest.
+
+    Args:
+        cache_metadata: Optional existing cache metadata to attach.
+        contents_count: Number of user contents to generate.
+        message_text: When provided, every content uses this exact text instead
+            of the default ``f"Test message {i}"``. Useful for building a
+            cacheable prefix large enough to clear Gemini's token minimum.
+    """
     contents = []
     for i in range(contents_count):
-      contents.append(
-          types.Content(
-              role="user", parts=[types.Part(text=f"Test message {i}")]
-          )
-      )
+      text = message_text if message_text is not None else f"Test message {i}"
+      contents.append(types.Content(role="user", parts=[types.Part(text=text)]))
 
     # Create tools for testing fingerprinting
     tools = [
@@ -202,6 +209,43 @@ async def test_handle_context_caching_invalid_cache_fingerprint_match(self):
     mock_cleanup.assert_called_once_with(existing_cache.cache_name)
     self.manager.genai_client.aio.caches.create.assert_called_once()
 
+  async def test_create_cache_gates_on_prefix_not_full_prompt(self):
+    """Cache creation is gated on the cacheable prefix, not the full prompt.
+
+    Regression test for https://github.com/google/adk-python/issues/5847.
+
+    On a long conversation the previous-prompt token count
+    (``cacheable_contents_token_count``) can be well above Gemini's 4096-token
+    minimum while the cached prefix ``contents[:cache_contents_count]`` is far
+    below it. Creating a cache in that case makes ``caches.create`` fail with a
+    400 INVALID_ARGUMENT. The manager must skip cache creation instead.
+    """
+    self.manager.genai_client.aio.caches.create = AsyncMock()
+
+    # A tiny cacheable prefix followed by a huge trailing user turn.
+    contents = [
+        types.Content(role="user", parts=[types.Part(text="Short prefix.")]),
+        types.Content(role="user", parts=[types.Part(text="word " * 100_000)]),
+    ]
+    llm_request = LlmRequest(
+        model="gemini-2.5-flash",
+        contents=contents,
+        config=types.GenerateContentConfig(
+            system_instruction="You are a helpful assistant.",
+        ),
+        cache_config=self.cache_config,
+    )
+    # Full previous prompt is large (clears the old, buggy gate)...
+    llm_request.cacheable_contents_token_count = 75000
+
+    # ...but only the tiny first content is cacheable.
+    result = await self.manager._create_new_cache_with_contents(
+        llm_request, cache_contents_count=1
+    )
+
+    assert result is None
+    self.manager.genai_client.aio.caches.create.assert_not_called()
+
   async def test_handle_context_caching_invalid_cache_fingerprint_mismatch(
       self,
   ):
@@ -904,8 +948,14 @@ async def test_fingerprint_only_metadata_transitions_to_active_cache(
     1. First call: no metadata -> returns fingerprint-only metadata
     2. Second call: fingerprint matches, cache created successfully
     """
+    # Use a prefix large enough to clear Gemini's token minimum, otherwise the
+    # cached prefix (contents[:3]) would be correctly rejected as too small.
+    big_text = "lorem ipsum " * 1000  # ~12k chars -> ~3k tokens per content
+
     # --- First LLM call: no existing metadata ---
-    llm_request_1 = self.create_llm_request(contents_count=3)
+    llm_request_1 = self.create_llm_request(
+        contents_count=3, message_text=big_text
+    )
 
     result_1 = await self.manager.handle_context_caching(llm_request_1)
 
@@ -916,9 +966,11 @@ async def test_fingerprint_only_metadata_transitions_to_active_cache(
     # --- Second LLM call: carry forward fingerprint-only metadata ---
     # Contents grew but we still have same prefix
     llm_request_2 = self.create_llm_request(
-        cache_metadata=result_1, contents_count=5
+        cache_metadata=result_1, contents_count=5, message_text=big_text
     )
-    llm_request_2.cacheable_contents_token_count = 4096
+    # Full previous prompt is large; the cached prefix (first 3 contents) also
+    # clears Gemini's 4096-token minimum.
+    llm_request_2.cacheable_contents_token_count = 30000
 
     # Verify prefix fingerprint matches (real implementation).
     # The fingerprint-only metadata is "invalid" (no cache_name),