fix: assemble only the current turn

The Interactions adapter already sends only the current turn when a previous interaction id is present, but the standard flow still walked and copied the whole session history first, and could invoke the explicit cache manager even though Interactions only supports implicit caching. This finds the previous interaction id up front, assembles just the current turn for those requests, and skips explicit cache mutation on that path. Ordinary GenerateContent requests keep full history.

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

Author: GWeale

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

@@ -90,7 +90,10 @@ async def run_async(
     )
 
     is_single_turn = getattr(agent, 'mode', None) == 'single_turn'
-    if agent.include_contents == 'default':
+    if (
+        agent.include_contents == 'default'
+        and not llm_request.previous_interaction_id
+    ):
       # Include full conversation history
       llm_request.contents = _get_contents(
           invocation_context.branch,
@@ -103,7 +106,8 @@ async def run_async(
           include_thoughts_from_other_agents=include_thoughts_from_other_agents,
       )
     else:
-      # Include current turn context only (no conversation history)
+      # Include current turn context only (no conversation history). Stateful
+      # Interactions requests already retain earlier turns server-side.
       llm_request.contents = _get_current_turn_contents(
           invocation_context.branch,
           invocation_context.session.events,

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

@@ -85,7 +85,7 @@ class InteractionsRequestProcessor(BaseLlmRequestProcessor):
   This processor extracts the previous_interaction_id from session events
   to enable stateful conversation chaining via the Interactions API.
   The actual content filtering (retaining only latest user messages) is
-  done in the Gemini class when using the Interactions API.
+  done by the content request processor after this processor runs.
   """
 
   async def run_async(

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

@@ -52,13 +52,13 @@ def _create_request_processors() -> list[BaseLlmRequestProcessor]:
       # Compaction should run before contents so compacted events are reflected
       # in the model request context.
       compaction.request_processor,
+      # Extract the Interactions chain id before contents. Chained requests
+      # only need the current turn because the service retains prior state.
+      interactions_processor.request_processor,
       contents.request_processor,
       # Context cache processor sets up cache config and finds
       # existing cache metadata.
       context_cache_processor.request_processor,
-      # Interactions processor extracts previous_interaction_id for
-      # stateful conversations via the Interactions API.
-      interactions_processor.request_processor,
       # Some implementations of NL Planning mark planning contents
       # as thoughts in the post processor.  Since these need to be
       # unmarked, NL Planning should be after contents.

src/google/adk/models/google_llm.py

@@ -200,7 +200,7 @@ async def generate_content_async(
     # Handle context caching if configured
     cache_metadata = None
     cache_manager = None
-    if llm_request.cache_config:
+    if llm_request.cache_config and not self.use_interactions_api:
       from ..telemetry.tracing import tracer
       from .gemini_context_cache_manager import GeminiContextCacheManager
 

tests/unittests/flows/llm_flows/test_contents.py

@@ -88,6 +88,44 @@ async def test_include_contents_default_full_history():
   ]
 
 
+@pytest.mark.asyncio
+async def test_chained_interactions_builds_only_current_turn_contents():
+  """Stateful Interactions requests do not copy history the server retains."""
+  agent = Agent(
+      model="gemini-2.5-flash", name="test_agent", include_contents="default"
+  )
+  llm_request = LlmRequest(
+      model="gemini-2.5-flash", previous_interaction_id="interaction-1"
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent
+  )
+  invocation_context.session.events = [
+      Event(
+          invocation_id="inv1",
+          author="user",
+          content=types.UserContent("Historical message"),
+      ),
+      Event(
+          invocation_id="inv2",
+          author="test_agent",
+          content=types.ModelContent("Historical response"),
+      ),
+      Event(
+          invocation_id="inv3",
+          author="user",
+          content=types.UserContent("Current message"),
+      ),
+  ]
+
+  async for _ in contents.request_processor.run_async(
+      invocation_context, llm_request
+  ):
+    pass
+
+  assert llm_request.contents == [types.UserContent("Current message")]
+
+
 @pytest.mark.asyncio
 async def test_include_contents_none_current_turn_only():
   """Test that include_contents='none' includes only current turn context."""

tests/unittests/flows/llm_flows/test_interactions_processor.py

@@ -17,9 +17,10 @@
 from unittest.mock import MagicMock
 
 from google.adk.events.event import Event
+from google.adk.flows.llm_flows import contents
 from google.adk.flows.llm_flows import interactions_processor
+from google.adk.flows.llm_flows.single_flow import SingleFlow
 from google.genai import types
-import pytest
 
 
 class TestInteractionsRequestProcessor:
@@ -224,6 +225,18 @@ def test_is_event_in_branch_root_events_included(self):
     )
 
 
+def test_single_flow_extracts_interaction_state_before_contents():
+  """Chained requests expose their interaction ID to content assembly."""
+  flow = SingleFlow()
+
+  interactions_index = flow.request_processors.index(
+      interactions_processor.request_processor
+  )
+  contents_index = flow.request_processors.index(contents.request_processor)
+
+  assert interactions_index < contents_index
+
+
 def _evt(author: str, interaction_id: str | None, branch: str | None) -> Event:
   return Event(author=author, interaction_id=interaction_id, branch=branch)
 

tests/unittests/models/test_google_llm.py

@@ -2056,6 +2056,46 @@ async def mock_coro():
       assert second_arg.invocations_used == cache_metadata.invocations_used
 
 
+@pytest.mark.asyncio
+async def test_interactions_api_does_not_apply_explicit_cache(llm_request):
+  """Interactions requests use implicit caching without mutating the prompt."""
+  gemini = Gemini(model="gemini-2.5-flash", use_interactions_api=True)
+  llm_request.cache_config = ContextCacheConfig()
+  original_request = llm_request.model_copy(deep=True)
+
+  async def generate_via_interactions(_llm_request, _stream):
+    yield LlmResponse(
+        content=Content(
+            role="model", parts=[Part.from_text(text="interaction response")]
+        )
+    )
+
+  with (
+      mock.patch.object(gemini, "_preprocess_request", new=AsyncMock()),
+      mock.patch.object(
+          gemini,
+          "_generate_content_via_interactions",
+          new=generate_via_interactions,
+      ),
+      mock.patch(
+          "google.adk.models.gemini_context_cache_manager.GeminiContextCacheManager"
+      ) as cache_manager_class,
+  ):
+    responses = [
+        response
+        async for response in gemini.generate_content_async(llm_request)
+    ]
+
+  assert responses[0].content.parts[0].text == "interaction response"
+  assert llm_request.contents == original_request.contents
+  assert (
+      llm_request.config.system_instruction
+      == original_request.config.system_instruction
+  )
+  assert llm_request.config.cached_content is None
+  cache_manager_class.assert_not_called()
+
+
 def test_build_function_declaration_log():
   """Test that _build_function_declaration_log formats function declarations correctly."""
   # Test case 1: Function with parameters and response