fix: Create context cache only when prefix matches with previous request

PiperOrigin-RevId: 817468275

src/google/adk/flows/llm_flows/context_cache_processor.py

@@ -72,11 +72,9 @@ class ContextCacheRequestProcessor(BaseLlmRequestProcessor):
     if latest_cache_metadata:
       llm_request.cache_metadata = latest_cache_metadata
       logger.debug(
-          'Found cache metadata for agent %s: invocations_used=%d, '
-          'cached_contents=%d',
+          'Found cache metadata for agent %s: %s',
           agent.name,
-          latest_cache_metadata.invocations_used,
-          latest_cache_metadata.cached_contents_count,
+          latest_cache_metadata,
       )
 
     if previous_token_count is not None:
@@ -108,8 +106,10 @@ class ContextCacheRequestProcessor(BaseLlmRequestProcessor):
 
     Returns:
         Tuple of (cache_metadata, previous_prompt_token_count)
-        cache_metadata: Latest cache metadata with updated invocations_used if needed
-        previous_prompt_token_count: Most recent prompt token count from LLM response
+        cache_metadata: Latest cache metadata with invocations_used incremented
+            only if this is a different invocation and has active cache
+        previous_prompt_token_count: Most recent prompt token count from
+            LLM response
     """
     if not invocation_context.session or not invocation_context.session.events:
       return None, None
@@ -126,17 +126,21 @@ class ContextCacheRequestProcessor(BaseLlmRequestProcessor):
 
       # Look for cache metadata (only in actual LLM response events)
       if cache_metadata is None and event.cache_metadata is not None:
-        # Check if this is a different invocation - increment invocations_used
-        if event.invocation_id and event.invocation_id != current_invocation_id:
-          # Different invocation - increment invocations_used
+        # Check if this is a different invocation and has active cache
+        if (
+            event.invocation_id
+            and event.invocation_id != current_invocation_id
+            and event.cache_metadata.cache_name is not None
+        ):
+          # Different invocation with active cache - increment invocations_used
           cache_metadata = event.cache_metadata.model_copy(
               update={
                   'invocations_used': event.cache_metadata.invocations_used + 1
               }
           )
         else:
-          # Same invocation or no invocation_id - return as-is
-          cache_metadata = event.cache_metadata
+          # Same invocation or no active cache - return copy as-is
+          cache_metadata = event.cache_metadata.model_copy()
 
       # Look for previous prompt token count (from actual LLM response events)
       if (

src/google/adk/models/cache_metadata.py

@@ -26,19 +26,30 @@ class CacheMetadata(BaseModel):
   """Metadata for context cache associated with LLM responses.
 
   This class stores cache identification, usage tracking, and lifecycle
-  information for a particular cache instance.
+  information for a particular cache instance. It can be in two states:
+
+  1. Active cache state: cache_name is set, all fields populated
+  2. Fingerprint-only state: cache_name is None, only fingerprint and
+     contents_count are set for prefix matching
 
   Token counts (cached and total) are available in the LlmResponse.usage_metadata
   and should be accessed from there to avoid duplication.
 
   Attributes:
       cache_name: The full resource name of the cached content (e.g.,
-          'projects/123/locations/us-central1/cachedContents/456')
-      expire_time: Unix timestamp when the cache expires
-      fingerprint: Hash of agent configuration (instruction + tools + model)
-      invocations_used: Number of invocations this cache has been used for
-      cached_contents_count: Number of contents stored in this cache
-      created_at: Unix timestamp when the cache was created
+          'projects/123/locations/us-central1/cachedContents/456').
+          None when no active cache exists (fingerprint-only state).
+      expire_time: Unix timestamp when the cache expires. None when no
+          active cache exists.
+      fingerprint: Hash of cacheable contents (instruction + tools + contents).
+          Always present for prefix matching.
+      invocations_used: Number of invocations this cache has been used for.
+          None when no active cache exists.
+      contents_count: Number of contents. When active cache exists, this is
+          the count of cached contents. When no active cache exists, this is
+          the total count of contents in the request.
+      created_at: Unix timestamp when the cache was created. None when
+          no active cache exists.
   """
 
   model_config = ConfigDict(
@@ -46,45 +57,65 @@ class CacheMetadata(BaseModel):
       frozen=True,  # Cache metadata should be immutable
   )
 
-  cache_name: str = Field(
-      description="Full resource name of the cached content"
+  cache_name: Optional[str] = Field(
+      default=None,
+      description=(
+          "Full resource name of the cached content (None if no active cache)"
+      ),
   )
 
-  expire_time: float = Field(description="Unix timestamp when cache expires")
+  expire_time: Optional[float] = Field(
+      default=None,
+      description="Unix timestamp when cache expires (None if no active cache)",
+  )
 
   fingerprint: str = Field(
-      description="Hash of agent configuration used to detect changes"
+      description="Hash of cacheable contents used to detect changes"
   )
 
-  invocations_used: int = Field(
+  invocations_used: Optional[int] = Field(
+      default=None,
       ge=0,
-      description="Number of invocations this cache has been used for",
+      description=(
+          "Number of invocations this cache has been used for (None if no"
+          " active cache)"
+      ),
   )
 
-  cached_contents_count: int = Field(
+  contents_count: int = Field(
       ge=0,
-      description="Number of contents stored in this cache",
+      description=(
+          "Number of contents (cached contents when active cache exists, "
+          "total contents in request when no active cache)"
+      ),
   )
 
   created_at: Optional[float] = Field(
       default=None,
       description=(
-          "Unix timestamp when cache was created (None if reused existing)"
+          "Unix timestamp when cache was created (None if no active cache)"
       ),
   )
 
   @property
   def expire_soon(self) -> bool:
     """Check if the cache will expire soon (with 2-minute buffer)."""
+    if self.expire_time is None:
+      return False
     buffer_seconds = 120  # 2 minutes buffer for processing time
     return time.time() > (self.expire_time - buffer_seconds)
 
   def __str__(self) -> str:
     """String representation for logging and debugging."""
+    if self.cache_name is None:
+      return (
+          f"Fingerprint-only: {self.contents_count} contents, "
+          f"fingerprint={self.fingerprint[:8]}..."
+      )
     cache_id = self.cache_name.split("/")[-1]
     time_until_expiry_minutes = (self.expire_time - time.time()) / 60
     return (
         f"Cache {cache_id}: used {self.invocations_used} invocations, "
-        f"cached {self.cached_contents_count} contents, "
+        f"cached {self.contents_count} contents, "
         f"expires in {time_until_expiry_minutes:.1f}min"
     )

src/google/adk/models/gemini_context_cache_manager.py

@@ -69,11 +69,8 @@ class GeminiContextCacheManager:
     # Check if we have existing cache metadata and if it's valid
     if llm_request.cache_metadata:
       logger.debug(
-          "Found existing cache metadata: cache_name=%s, invocations_used=%d,"
-          " cached_contents_count=%d",
-          llm_request.cache_metadata.cache_name,
-          llm_request.cache_metadata.invocations_used,
-          llm_request.cache_metadata.cached_contents_count,
+          "Found existing cache metadata: %s",
+          llm_request.cache_metadata,
       )
       if await self._is_cache_valid(llm_request):
         # Valid cache found - use it
@@ -82,46 +79,69 @@ class GeminiContextCacheManager:
             llm_request.cache_metadata.cache_name,
         )
         cache_name = llm_request.cache_metadata.cache_name
-        cache_contents_count = llm_request.cache_metadata.cached_contents_count
+        cache_contents_count = llm_request.cache_metadata.contents_count
         self._apply_cache_to_request(
             llm_request, cache_name, cache_contents_count
         )
         return llm_request.cache_metadata.model_copy()
       else:
-        # Invalid cache - clean it up
-        logger.debug(
-            "Cache is invalid, cleaning up: %s",
-            llm_request.cache_metadata.cache_name,
+        # Invalid cache - clean it up and check if we should create new one
+        old_cache_metadata = llm_request.cache_metadata
+
+        # Only cleanup if there's an active cache
+        if old_cache_metadata.cache_name is not None:
+          logger.debug(
+              "Cache is invalid, cleaning up: %s",
+              old_cache_metadata.cache_name,
+          )
+          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
+        current_fingerprint = self._generate_cache_fingerprint(
+            llm_request, cache_contents_count
         )
-        await self.cleanup_cache(llm_request.cache_metadata.cache_name)
-        llm_request.cache_metadata = None
 
-    # Find contents to cache for new cache creation
-    cache_contents_count = self._find_count_of_contents_to_cache(
-        llm_request.contents
-    )
+        # 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"
+          )
+          cache_metadata = await self._create_new_cache_with_contents(
+              llm_request, cache_contents_count
+          )
+          if cache_metadata:
+            self._apply_cache_to_request(
+                llm_request, cache_metadata.cache_name, cache_contents_count
+            )
+            return cache_metadata
+
+        # Fingerprints don't match - recalculate with total contents
+        logger.debug(
+            "Fingerprints don't match, returning fingerprint-only metadata"
+        )
+        total_contents_count = len(llm_request.contents)
+        fingerprint_for_all = self._generate_cache_fingerprint(
+            llm_request, total_contents_count
+        )
+        return CacheMetadata(
+            fingerprint=fingerprint_for_all,
+            contents_count=total_contents_count,
+        )
+
+    # No existing cache metadata - return fingerprint-only metadata
+    # We don't create cache without previous fingerprint to match
     logger.debug(
-        "Determined to cache %d contents from %d total contents",
-        cache_contents_count,
-        len(llm_request.contents),
+        "No existing cache metadata, creating fingerprint-only metadata"
     )
-
-    # Create new cache with the determined contents
-    cache_metadata = await self._create_new_cache_with_contents(
-        llm_request, cache_contents_count
+    total_contents_count = len(llm_request.contents)
+    fingerprint = self._generate_cache_fingerprint(
+        llm_request, total_contents_count
     )
-    if not cache_metadata:
-      return None
-
-    # Set up request to use the new cache
-    self._apply_cache_to_request(
-        llm_request, cache_metadata.cache_name, cache_contents_count
+    return CacheMetadata(
+        fingerprint=fingerprint,
+        contents_count=total_contents_count,
     )
-    logger.debug(
-        "Successfully applied cache to request: %s", cache_metadata.cache_name
-    )
-
-    return cache_metadata
 
   def _find_count_of_contents_to_cache(
       self, contents: list[types.Content]
@@ -158,7 +178,8 @@ class GeminiContextCacheManager:
   async def _is_cache_valid(self, llm_request: LlmRequest) -> bool:
     """Check if the cache from request metadata is still valid.
 
-    Validates expiry, cache intervals, and fingerprint compatibility.
+    Validates that it's an active cache (not fingerprint-only), checks expiry,
+    cache intervals, and fingerprint compatibility.
 
     Args:
         llm_request: Request containing cache metadata to validate
@@ -170,6 +191,10 @@ class GeminiContextCacheManager:
     if not cache_metadata:
       return False
 
+    # Fingerprint-only metadata is not a valid active cache
+    if cache_metadata.cache_name is None:
+      return False
+
     # Check if cache has expired
     if time.time() >= cache_metadata.expire_time:
       logger.info("Cache expired: %s", cache_metadata.cache_name)
@@ -190,7 +215,7 @@ class GeminiContextCacheManager:
 
     # Check if fingerprint matches using cached contents count
     current_fingerprint = self._generate_cache_fingerprint(
-        llm_request, cache_metadata.cached_contents_count
+        llm_request, cache_metadata.contents_count
     )
     if current_fingerprint != cache_metadata.fingerprint:
       logger.debug("Cache content fingerprint mismatch")
@@ -376,7 +401,7 @@ class GeminiContextCacheManager:
             llm_request, cache_contents_count
         ),
         invocations_used=1,
-        cached_contents_count=cache_contents_count,
+        contents_count=cache_contents_count,
         created_at=created_at,
     )
 

tests/unittests/agents/test_gemini_context_cache_manager.py

@@ -87,7 +87,7 @@ class TestGeminiContextCacheManager:
     )
 
   def create_cache_metadata(
-      self, invocations_used=0, expired=False, cached_contents_count=3
+      self, invocations_used=0, expired=False, contents_count=3
   ):
     """Helper to create test CacheMetadata."""
     current_time = time.time()
@@ -98,7 +98,7 @@ class TestGeminiContextCacheManager:
         expire_time=expire_time,
         fingerprint="test_fingerprint",
         invocations_used=invocations_used,
-        cached_contents_count=cached_contents_count,
+        contents_count=contents_count,
         created_at=current_time - 600,
     )
 
@@ -109,45 +109,26 @@ class TestGeminiContextCacheManager:
     assert manager is not None
     assert manager.genai_client == mock_client
 
-  async def test_handle_context_caching_new_cache(self):
-    """Test handling context caching with no existing cache."""
-    # Setup mocks
-    mock_cached_content = AsyncMock()
-    mock_cached_content.name = (
-        "projects/test/locations/us-central1/cachedContents/new123"
-    )
-    self.manager.genai_client.aio.caches.create = AsyncMock(
-        return_value=mock_cached_content
-    )
-
-    llm_request = self.create_llm_request()
-    llm_request.cacheable_contents_token_count = (
-        2048  # Add token count for cache creation
-    )
-    start_time = time.time()
+  async def test_handle_context_caching_no_existing_cache(self):
+    """Test handling context caching with no existing cache returns fingerprint-only metadata."""
+    llm_request = self.create_llm_request(contents_count=5)
 
     with patch.object(
         self.manager, "_generate_cache_fingerprint", return_value="test_fp"
     ):
       result = await self.manager.handle_context_caching(llm_request)
 
-    end_time = time.time()
-
     assert result is not None
-    # Verify new cache metadata is created with fresh values
-    assert (
-        result.cache_name
-        == "projects/test/locations/us-central1/cachedContents/new123"
-    )
-    assert result.invocations_used == 1  # New cache starts with 1 invocation
+    # Should return fingerprint-only metadata (no active cache)
+    assert result.cache_name is None
+    assert result.expire_time is None
+    assert result.invocations_used is None
+    assert result.created_at is None
     assert result.fingerprint == "test_fp"
+    assert result.contents_count == 5  # Total contents count
 
-    # Verify timestamps are recent (within test execution time)
-    assert start_time <= result.created_at <= end_time
-    assert result.expire_time > time.time()  # Should be in the future
-
-    # Verify cache creation was called
-    self.manager.genai_client.aio.caches.create.assert_called_once()
+    # No cache should be created
+    self.manager.genai_client.aio.caches.create.assert_not_called()
 
   async def test_handle_context_caching_valid_existing_cache(self):
     """Test handling context caching with valid existing cache."""
@@ -181,8 +162,8 @@ class TestGeminiContextCacheManager:
     # Should not create new cache
     self.manager.genai_client.aio.caches.create.assert_not_called()
 
-  async def test_handle_context_caching_invalid_existing_cache(self):
-    """Test handling context caching with invalid existing cache."""
+  async def test_handle_context_caching_invalid_cache_fingerprint_match(self):
+    """Test invalid cache with matching fingerprint creates new cache."""
     # Setup mocks
     mock_cached_content = AsyncMock()
     mock_cached_content.name = (
@@ -205,13 +186,16 @@ class TestGeminiContextCacheManager:
         patch.object(self.manager, "_is_cache_valid", return_value=False),
         patch.object(self.manager, "cleanup_cache") as mock_cleanup,
         patch.object(
-            self.manager, "_generate_cache_fingerprint", return_value="new_fp"
+            self.manager,
+            "_generate_cache_fingerprint",
+            return_value="test_fingerprint",  # Match old fingerprint
         ),
     ):
 
       result = await self.manager.handle_context_caching(llm_request)
 
     assert result is not None
+    # Should create new cache when fingerprints match
     assert (
         result.cache_name
         == "projects/test/locations/us-central1/cachedContents/new456"
@@ -219,6 +203,41 @@ class TestGeminiContextCacheManager:
     mock_cleanup.assert_called_once_with(existing_cache.cache_name)
     self.manager.genai_client.aio.caches.create.assert_called_once()
 
+  async def test_handle_context_caching_invalid_cache_fingerprint_mismatch(
+      self,
+  ):
+    """Test invalid cache with mismatched fingerprint returns fingerprint-only metadata."""
+    # Create request with invalid existing cache
+    existing_cache = self.create_cache_metadata(
+        invocations_used=15, contents_count=3
+    )  # Exceeds cache_intervals
+    llm_request = self.create_llm_request(
+        cache_metadata=existing_cache, contents_count=5
+    )
+
+    with (
+        patch.object(self.manager, "_is_cache_valid", return_value=False),
+        patch.object(self.manager, "cleanup_cache") as mock_cleanup,
+        patch.object(
+            self.manager,
+            "_generate_cache_fingerprint",
+            side_effect=["old_fp", "new_fp"],  # Different fingerprints
+        ),
+    ):
+
+      result = await self.manager.handle_context_caching(llm_request)
+
+    assert result is not None
+    # Should return fingerprint-only metadata
+    assert result.cache_name is None
+    assert result.expire_time is None
+    assert result.invocations_used is None
+    assert result.created_at is None
+    assert result.fingerprint == "new_fp"
+    assert result.contents_count == 5  # Total contents count
+    mock_cleanup.assert_called_once_with(existing_cache.cache_name)
+    self.manager.genai_client.aio.caches.create.assert_not_called()
+
   async def test_is_cache_valid_fingerprint_mismatch(self):
     """Test cache validation with fingerprint mismatch."""
     cache_metadata = self.create_cache_metadata()
@@ -247,6 +266,21 @@ class TestGeminiContextCacheManager:
 
     assert result is False
 
+  async def test_is_cache_valid_fingerprint_only_metadata(self):
+    """Test cache validation with fingerprint-only metadata (no active cache)."""
+    # Create fingerprint-only metadata (cache_name is None)
+    cache_metadata = CacheMetadata(
+        fingerprint="test_fingerprint",
+        contents_count=5,
+    )
+    llm_request = self.create_llm_request(cache_metadata=cache_metadata)
+
+    result = await self.manager._is_cache_valid(llm_request)
+
+    assert (
+        result is False
+    )  # Fingerprint-only metadata is not a valid active cache
+
   async def test_is_cache_valid_cache_intervals_exceeded(self):
     """Test cache validation with max invocations exceeded."""
     cache_metadata = self.create_cache_metadata(
@@ -537,16 +571,8 @@ class TestGeminiContextCacheManager:
     return llm_request
 
   async def test_cache_creation_with_sufficient_token_count(self):
-    """Test cache creation succeeds when token count meets minimum."""
-    # Setup mocks
-    mock_cached_content = AsyncMock()
-    mock_cached_content.name = (
-        "projects/test/locations/us-central1/cachedContents/token123"
-    )
-    self.manager.genai_client.aio.caches.create = AsyncMock(
-        return_value=mock_cached_content
-    )
-
+    """Test that fingerprint-only metadata is returned even with sufficient tokens."""
+    # With new prefix matching logic, no cache is created without existing metadata
     # Create request with sufficient token count
     llm_request = self.create_llm_request_with_token_count(token_count=2048)
 
@@ -555,13 +581,15 @@ class TestGeminiContextCacheManager:
     ):
       result = await self.manager.handle_context_caching(llm_request)
 
-    # Should succeed in creating cache
+    # Should return fingerprint-only metadata (no cache creation)
     assert result is not None
-    assert result.cache_name == mock_cached_content.name
-    self.manager.genai_client.aio.caches.create.assert_called_once()
+    assert result.cache_name is None  # Fingerprint-only state
+    assert result.fingerprint == "test_fp"
+    assert result.contents_count == 3
+    self.manager.genai_client.aio.caches.create.assert_not_called()
 
   async def test_cache_creation_with_insufficient_token_count(self):
-    """Test cache creation fails when token count is below minimum."""
+    """Test that fingerprint-only metadata is returned even with insufficient tokens."""
     # Set higher minimum token requirement
     self.manager.cache_config = ContextCacheConfig(
         cache_intervals=10,
@@ -573,19 +601,29 @@ class TestGeminiContextCacheManager:
     llm_request = self.create_llm_request_with_token_count(token_count=1024)
     llm_request.cache_config = self.manager.cache_config
 
-    result = await self.manager.handle_context_caching(llm_request)
+    with patch.object(
+        self.manager, "_generate_cache_fingerprint", return_value="test_fp"
+    ):
+      result = await self.manager.handle_context_caching(llm_request)
 
-    # Should not create cache
-    assert result is None
+    # Should return fingerprint-only metadata
+    assert result is not None
+    assert result.cache_name is None
+    assert result.fingerprint == "test_fp"
     self.manager.genai_client.aio.caches.create.assert_not_called()
 
   async def test_cache_creation_without_token_count(self):
-    """Test cache creation is skipped when no token count is available."""
+    """Test that fingerprint-only metadata is returned even without token count."""
     # Create request without token count (initial request)
     llm_request = self.create_llm_request_with_token_count(token_count=None)
 
-    result = await self.manager.handle_context_caching(llm_request)
+    with patch.object(
+        self.manager, "_generate_cache_fingerprint", return_value="test_fp"
+    ):
+      result = await self.manager.handle_context_caching(llm_request)
 
-    # Should skip cache creation for initial request
-    assert result is None
+    # Should return fingerprint-only metadata
+    assert result is not None
+    assert result.cache_name is None
+    assert result.fingerprint == "test_fp"
     self.manager.genai_client.aio.caches.create.assert_not_called()

tests/unittests/flows/llm_flows/test_context_cache_processor.py

@@ -66,7 +66,7 @@ class TestContextCacheRequestProcessor:
     )
 
   def create_cache_metadata(
-      self, invocations_used=1, cache_name="test-cache", cached_contents_count=3
+      self, invocations_used=1, cache_name="test-cache", contents_count=3
   ):
     """Helper to create CacheMetadata."""
     return CacheMetadata(
@@ -76,7 +76,7 @@ class TestContextCacheRequestProcessor:
         expire_time=time.time() + 1800,
         fingerprint="test_fingerprint",
         invocations_used=invocations_used,
-        cached_contents_count=cached_contents_count,
+        contents_count=contents_count,
         created_at=time.time() - 600,
     )
 

tests/unittests/models/test_cache_metadata.py

@@ -32,7 +32,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=5,
-        cached_contents_count=3,
+        contents_count=3,
     )
 
     assert (
@@ -42,7 +42,7 @@ class TestCacheMetadata:
     assert metadata.expire_time > time.time()
     assert metadata.fingerprint == "abc123"
     assert metadata.invocations_used == 5
-    assert metadata.cached_contents_count == 3
+    assert metadata.contents_count == 3
     assert metadata.created_at is None  # Optional field
 
   def test_optional_created_at(self):
@@ -54,7 +54,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=3,
-        cached_contents_count=2,
+        contents_count=2,
         created_at=current_time,
     )
 
@@ -68,7 +68,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=0,
-        cached_contents_count=1,
+        contents_count=1,
     )
     assert metadata.invocations_used == 0
 
@@ -77,7 +77,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=10,
-        cached_contents_count=1,
+        contents_count=1,
     )
     assert metadata.invocations_used == 10
 
@@ -88,30 +88,30 @@ class TestCacheMetadata:
           expire_time=time.time() + 1800,
           fingerprint="abc123",
           invocations_used=-1,
-          cached_contents_count=1,
+          contents_count=1,
       )
     assert "greater than or equal to 0" in str(exc_info.value)
 
-  def test_cached_contents_count_validation(self):
-    """Test cached_contents_count validation constraints."""
+  def test_contents_count_validation(self):
+    """Test contents_count validation constraints."""
     # Valid: zero or positive
     metadata = CacheMetadata(
         cache_name="projects/123/locations/us-central1/cachedContents/456",
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=1,
-        cached_contents_count=0,
+        contents_count=0,
     )
-    assert metadata.cached_contents_count == 0
+    assert metadata.contents_count == 0
 
     metadata = CacheMetadata(
         cache_name="projects/123/locations/us-central1/cachedContents/456",
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=1,
-        cached_contents_count=10,
+        contents_count=10,
     )
-    assert metadata.cached_contents_count == 10
+    assert metadata.contents_count == 10
 
     # Invalid: negative
     with pytest.raises(ValidationError) as exc_info:
@@ -120,7 +120,7 @@ class TestCacheMetadata:
           expire_time=time.time() + 1800,
           fingerprint="abc123",
           invocations_used=1,
-          cached_contents_count=-1,
+          contents_count=-1,
       )
     assert "greater than or equal to 0" in str(exc_info.value)
 
@@ -133,7 +133,7 @@ class TestCacheMetadata:
         expire_time=future_time,
         fingerprint="abc123",
         invocations_used=1,
-        cached_contents_count=1,
+        contents_count=1,
     )
     assert not metadata.expire_soon
 
@@ -144,7 +144,7 @@ class TestCacheMetadata:
         expire_time=soon_time,
         fingerprint="abc123",
         invocations_used=1,
-        cached_contents_count=1,
+        contents_count=1,
     )
     assert metadata.expire_soon
 
@@ -158,7 +158,7 @@ class TestCacheMetadata:
         expire_time=expire_time,
         fingerprint="abc123",
         invocations_used=7,
-        cached_contents_count=4,
+        contents_count=4,
     )
 
     str_repr = str(metadata)
@@ -174,7 +174,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=5,
-        cached_contents_count=3,
+        contents_count=3,
     )
 
     # Should not be able to modify fields
@@ -188,7 +188,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=5,
-        cached_contents_count=3,
+        contents_count=3,
     )
 
     assert metadata.model_config["extra"] == "forbid"
@@ -201,7 +201,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=5,
-        cached_contents_count=3,
+        contents_count=3,
     )
     schema = metadata.model_json_schema()
 
@@ -211,10 +211,10 @@ class TestCacheMetadata:
         in schema["properties"]["invocations_used"]["description"]
     )
 
-    assert "cached_contents_count" in schema["properties"]
+    assert "contents_count" in schema["properties"]
     assert (
         "Number of contents"
-        in schema["properties"]["cached_contents_count"]["description"]
+        in schema["properties"]["contents_count"]["description"]
     )
 
   def test_realistic_cache_scenarios(self):
@@ -227,7 +227,7 @@ class TestCacheMetadata:
         expire_time=current_time + 1800,
         fingerprint="fresh_fingerprint",
         invocations_used=1,
-        cached_contents_count=5,
+        contents_count=5,
         created_at=current_time,
     )
     assert fresh_cache.invocations_used == 1
@@ -239,7 +239,7 @@ class TestCacheMetadata:
         expire_time=current_time + 600,
         fingerprint="used_fingerprint",
         invocations_used=8,
-        cached_contents_count=3,
+        contents_count=3,
         created_at=current_time - 1200,
     )
     assert used_cache.invocations_used == 8
@@ -252,7 +252,7 @@ class TestCacheMetadata:
         expire_time=current_time + 60,  # 1 minute
         fingerprint="expiring_fingerprint",
         invocations_used=15,
-        cached_contents_count=10,
+        contents_count=10,
     )
     assert expiring_cache.expire_soon
 
@@ -265,7 +265,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=1,
-        cached_contents_count=2,
+        contents_count=2,
     )
 
     str_repr = str(metadata)
@@ -278,7 +278,7 @@ class TestCacheMetadata:
         expire_time=time.time() + 1800,
         fingerprint="abc123",
         invocations_used=5,
-        cached_contents_count=3,
+        contents_count=3,
     )
 
     # Verify that token counts are NOT in CacheMetadata
@@ -288,22 +288,17 @@ class TestCacheMetadata:
     assert not hasattr(metadata, "prompt_tokens")
 
   def test_missing_required_fields(self):
-    """Test validation when required fields are missing."""
-    # Test each required field
+    """Test validation when truly required fields are missing."""
+    # Only fingerprint and contents_count are required now
+    # Other fields are optional (for fingerprint-only state)
     required_fields = [
-        "cache_name",
-        "expire_time",
         "fingerprint",
-        "invocations_used",
-        "cached_contents_count",
+        "contents_count",
     ]
 
     base_args = {
-        "cache_name": "projects/123/locations/us-central1/cachedContents/456",
-        "expire_time": time.time() + 1800,
         "fingerprint": "abc123",
-        "invocations_used": 1,
-        "cached_contents_count": 2,
+        "contents_count": 2,
     }
 
     for field in required_fields:
@@ -312,3 +307,13 @@ class TestCacheMetadata:
 
       with pytest.raises(ValidationError):
         CacheMetadata(**args)
+
+    # Test that optional fields can be omitted (fingerprint-only state)
+    metadata = CacheMetadata(
+        fingerprint="abc123",
+        contents_count=5,
+    )
+    assert metadata.cache_name is None
+    assert metadata.expire_time is None
+    assert metadata.invocations_used is None
+    assert metadata.created_at is None

tests/unittests/models/test_google_llm.py

@@ -96,7 +96,7 @@ def cache_metadata():
       expire_time=time.time() + 3600,
       fingerprint="test_fingerprint",
       invocations_used=2,
-      cached_contents_count=3,
+      contents_count=3,
       created_at=time.time() - 600,
   )
 

tests/unittests/utils/test_cache_performance_analyzer.py

@@ -36,7 +36,7 @@ class TestCachePerformanceAnalyzer:
     self.analyzer = CachePerformanceAnalyzer(self.mock_session_service)
 
   def create_cache_metadata(
-      self, invocations_used=1, cache_name="test-cache", cached_contents_count=5
+      self, invocations_used=1, cache_name="test-cache", contents_count=5
   ):
     """Helper to create test CacheMetadata."""
     return CacheMetadata(
@@ -46,7 +46,7 @@ class TestCachePerformanceAnalyzer:
         expire_time=time.time() + 1800,
         fingerprint="test_fingerprint",
         invocations_used=invocations_used,
-        cached_contents_count=cached_contents_count,
+        contents_count=contents_count,
         created_at=time.time() - 600,
     )