feat: record implicit vs explicit context cache type in analytics

Cached token counts alone cannot distinguish Gemini provider-side
implicit prefix caching from ADK-managed explicit CachedContent. Derive a
cache_type (explicit/implicit/none) on the final response and expose it
in the BigQuery analytics view so the two can be reported separately.

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

Author: GWeale

src/google/adk/plugins/bigquery_agent_analytics_plugin.py

@@ -3655,6 +3655,9 @@ def _parse_custom_metadata_allowlist(
         "JSON_VALUE(attributes, '$.model_version') AS model_version",
         "JSON_QUERY(attributes, '$.usage_metadata') AS usage_metadata",
         "JSON_QUERY(attributes, '$.cache_metadata') AS cache_metadata",
+        # NULL on partial streaming rows and pre-CL rows; filter to final
+        # responses before aggregating on cache_type.
+        "JSON_VALUE(attributes, '$.cache_type') AS cache_type",
     ],
     "LLM_ERROR": [
         "CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) AS total_ms",
@@ -6761,6 +6764,9 @@ async def after_model_callback(
     is_popped = False
     duration = 0
     tfft = None
+    extra_attributes: dict[str, Any] = {}
+    usage_metadata = llm_response.usage_metadata
+    cache_metadata = getattr(llm_response, "cache_metadata", None)
 
     if hasattr(llm_response, "partial") and llm_response.partial:
       # Streaming chunk - do NOT pop span yet
@@ -6796,6 +6802,29 @@ async def after_model_callback(
       # Otherwise log_event will fetch current stack (which is parent).
       span_id = popped_span_id or span_id
 
+      # cache_type classifies the cached-token hit so analytics can separate
+      # ADK-managed explicit caching from Gemini provider-side implicit prefix
+      # caching (the two are indistinguishable from token counts alone).
+      # cache_metadata is attached only when ADK explicit caching is configured,
+      # so its presence means explicit (including the fingerprint-only,
+      # cache_name=None state). No cached tokens -> "none", regardless of
+      # whether a cache is configured. Only derived on the final response.
+      # Token counts carry no source, so when ADK caching is configured the
+      # cache_metadata presence wins the "explicit" label even if some of the
+      # cached tokens came from provider-side implicit caching.
+      cached = bool(
+          usage_metadata
+          and (getattr(usage_metadata, "cached_content_token_count", 0) or 0)
+          > 0
+      )
+      if not cached:
+        cache_type = "none"
+      elif cache_metadata is not None:
+        cache_type = "explicit"
+      else:
+        cache_type = "implicit"
+      extra_attributes["cache_type"] = cache_type
+
     await self._log_event(
         "LLM_RESPONSE",
         callback_context,
@@ -6805,10 +6834,11 @@ async def after_model_callback(
             latency_ms=duration,
             time_to_first_token_ms=tfft,
             model_version=llm_response.model_version,
-            usage_metadata=llm_response.usage_metadata,
-            cache_metadata=getattr(llm_response, "cache_metadata", None),
+            usage_metadata=usage_metadata,
+            cache_metadata=cache_metadata,
             span_id_override=span_id if is_popped else None,
             parent_span_id_override=(parent_span_id if is_popped else None),
+            extra_attributes=extra_attributes,
         ),
     )
 

tests/unittests/plugins/test_bigquery_agent_analytics_plugin.py

@@ -8088,6 +8088,203 @@ def __init__(self):
     attributes = json.loads(log_entry["attributes"])
     assert "cache_metadata" not in attributes
 
+  async def _run_after_model(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+      llm_response,
+  ):
+    """Drives after_model_callback and returns the LLM_RESPONSE attributes."""
+    bigquery_agent_analytics_plugin.TraceManager.push_span(callback_context)
+    await bq_plugin_inst.after_model_callback(
+        callback_context=callback_context,
+        llm_response=llm_response,
+    )
+    await asyncio.sleep(0.05)
+    rows = await _get_captured_rows_async(mock_write_client, dummy_arrow_schema)
+    log_entry = next(r for r in rows if r["event_type"] == "LLM_RESPONSE")
+    return json.loads(log_entry["attributes"])
+
+  @pytest.mark.asyncio
+  async def test_cache_type_explicit(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """cache_name set + cached tokens -> explicit (ADK-managed cache)."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+            cached_content_token_count=80,
+        ),
+        cache_metadata={
+            "cache_name": "projects/p/locations/us-central1/cachedContents/c",
+            "expire_time": 9999999999.0,
+            "fingerprint": "fp-1",
+            "invocations_used": 1,
+            "contents_count": 2,
+            "created_at": 1.0,
+        },
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert attributes["cache_type"] == "explicit"
+
+  @pytest.mark.asyncio
+  async def test_cache_type_implicit(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """Cached tokens with no cache_metadata -> implicit (provider prefix)."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+            cached_content_token_count=80,
+        ),
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert attributes["cache_type"] == "implicit"
+
+  @pytest.mark.asyncio
+  async def test_cache_type_explicit_fingerprint_only(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """Fingerprint-only cache_metadata (cache_name=None) is still explicit."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+            cached_content_token_count=80,
+        ),
+        cache_metadata={"fingerprint": "fp-1", "contents_count": 2},
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert attributes["cache_type"] == "explicit"
+
+  @pytest.mark.asyncio
+  async def test_cache_type_none_with_active_cache(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """Active cache but no cached tokens (creation turn / miss) -> none."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+        ),
+        cache_metadata={
+            "cache_name": "projects/p/locations/us-central1/cachedContents/c",
+            "expire_time": 9999999999.0,
+            "fingerprint": "fp-1",
+            "invocations_used": 1,
+            "contents_count": 2,
+            "created_at": 1.0,
+        },
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert attributes["cache_type"] == "none"
+
+  @pytest.mark.asyncio
+  async def test_cache_type_none(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """No cached tokens -> none."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+        ),
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert attributes["cache_type"] == "none"
+
+  @pytest.mark.asyncio
+  async def test_cache_type_absent_on_partial_response(
+      self,
+      bq_plugin_inst,
+      mock_write_client,
+      callback_context,
+      dummy_arrow_schema,
+  ):
+    """Partial streaming rows carry no cache_type, even with cached tokens."""
+    llm_response = llm_response_lib.LlmResponse(
+        content=types.Content(parts=[types.Part(text="hi")]),
+        partial=True,
+        usage_metadata=types.GenerateContentResponseUsageMetadata(
+            prompt_token_count=100,
+            candidates_token_count=20,
+            total_token_count=120,
+            cached_content_token_count=80,
+        ),
+    )
+    attributes = await self._run_after_model(
+        bq_plugin_inst,
+        mock_write_client,
+        callback_context,
+        dummy_arrow_schema,
+        llm_response,
+    )
+    assert "cache_type" not in attributes
+
 
 # ==============================================================
 # TEST CLASS: A2A_INTERACTION event logging via on_event_callback