fix(caching): prevent prompt cache invalidation when using dynamic tools

Introduce a framework-level `_dynamic_instructions` pipeline in ADK to prevent prompt cache invalidation when tools inject dynamic context (artifacts, memory preloads, MCP resources).

Instead of tools eagerly appending instructions to `system_instruction` during `process_llm_request` (which alters the static system prompt prefix and invalidates Gemini prompt caches on every turn), tools now register instructions via `llm_request._append_dynamic_instructions`.

A finalizer (`_finalize_dynamic_instructions`) in `base_llm_flow.py` resolves these dynamic instructions post-tool execution.

To maintain backward compatibility for existing agents, this behavior is gated behind the `DYNAMIC_INSTRUCTION_ROUTING` feature flag:
- When enabled, dynamic instructions are routed into `contents` (user turn) via `_add_instructions_to_user_content`, preserving the static system prompt prefix for caching.
- When disabled (default), dynamic instructions append to `system_instruction`, preserving legacy prompt behavior.

To enable this feature:
- Environment variable: `export ADK_ENABLE_DYNAMIC_INSTRUCTION_ROUTING=1`
- Python code: `override_feature_enabled(FeatureName.DYNAMIC_INSTRUCTION_ROUTING, True)`
- ADK CLI: `adk web --enable_features=DYNAMIC_INSTRUCTION_ROUTING`

Closes #3227

Co-authored-by: Yi Liu <yiliuly@google.com>
PiperOrigin-RevId: 951562128

Author: i-yliu

src/google/adk/features/_feature_registry.py

@@ -37,6 +37,7 @@ class FeatureName(str, Enum):
   COMPUTER_USE = "COMPUTER_USE"
   DATA_AGENT_TOOL_CONFIG = "DATA_AGENT_TOOL_CONFIG"
   DATA_AGENT_TOOLSET = "DATA_AGENT_TOOLSET"
+  DYNAMIC_INSTRUCTION_ROUTING = "DYNAMIC_INSTRUCTION_ROUTING"
   DAYTONA_ENVIRONMENT = "DAYTONA_ENVIRONMENT"
   E2B_ENVIRONMENT = "E2B_ENVIRONMENT"
   ENVIRONMENT_SIMULATION = "ENVIRONMENT_SIMULATION"
@@ -131,6 +132,9 @@ class FeatureConfig:
     FeatureName.DATA_AGENT_TOOLSET: FeatureConfig(
         FeatureStage.EXPERIMENTAL, default_on=True
     ),
+    FeatureName.DYNAMIC_INSTRUCTION_ROUTING: FeatureConfig(
+        FeatureStage.EXPERIMENTAL, default_on=False
+    ),
     FeatureName.DAYTONA_ENVIRONMENT: FeatureConfig(
         FeatureStage.EXPERIMENTAL, default_on=True
     ),

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

@@ -1091,6 +1091,9 @@ async def _preprocess_async(
     # Run processors for tools.
     await _process_agent_tools(invocation_context, llm_request)
 
+    # Finalize dynamic instructions from tools.
+    await _finalize_dynamic_instructions(invocation_context, llm_request)
+
   async def _postprocess_async(
       self,
       invocation_context: InvocationContext,
@@ -1614,3 +1617,36 @@ def __get_llm(self, invocation_context: InvocationContext) -> BaseLlm:
           f' but got {type(agent)}'
       )
     return agent.canonical_model
+
+
+async def _finalize_dynamic_instructions(
+    invocation_context: InvocationContext,
+    llm_request: LlmRequest,
+) -> None:
+  """Finalizes and resolves dynamic instructions from LlmRequest."""
+  if not llm_request._dynamic_instructions:
+    return
+
+  combined_text = '\n\n'.join(llm_request._dynamic_instructions)
+
+  from ...features import FeatureName
+  from ...features import is_feature_enabled
+
+  # TODO: Deprecate system_instruction fallback and make user content routing standard.
+  if is_feature_enabled(FeatureName.DYNAMIC_INSTRUCTION_ROUTING):
+    from .contents import _add_instructions_to_user_content
+
+    instruction_content = types.Content(
+        role='user',
+        parts=[types.Part.from_text(text=combined_text)],
+    )
+    await _add_instructions_to_user_content(
+        invocation_context,
+        llm_request,
+        [instruction_content],
+    )
+  else:
+    llm_request.append_instructions([combined_text])
+
+  # Clear dynamic instructions to prevent double finalization.
+  llm_request._dynamic_instructions.clear()

src/google/adk/models/llm_request.py

@@ -101,6 +101,9 @@ class LlmRequest(BaseModel):
   the full history.
   """
 
+  _dynamic_instructions: list[str] = PrivateAttr(default_factory=list)
+  """Dynamic instructions generated by tools that should be resolved in finalized contents/system instructions."""
+
   _is_managed_agent: bool = PrivateAttr(default=False)
   """Internal flag: whether this request was built by a ManagedAgent.
 
@@ -110,6 +113,10 @@ class LlmRequest(BaseModel):
   server-side without a Gemini model.
   """
 
+  def _append_dynamic_instructions(self, instructions: list[str]) -> None:
+    """Appends dynamic instructions to the request."""
+    self._dynamic_instructions.extend(instructions)
+
   def append_instructions(
       self, instructions: Union[list[str], types.Content]
   ) -> list[types.Content]:

src/google/adk/tools/load_artifacts_tool.py

@@ -272,16 +272,16 @@ async def _append_artifacts_to_llm_request(
     if not artifact_names:
       return
 
-    # Tell the model about the available artifacts.
-    llm_request.append_instructions([f"""You have a list of artifacts:
+    instruction_text = f"""You have a list of artifacts:
   {json.dumps(artifact_names)}
 
   When the user asks questions about any of the artifacts, you should call the
   `load_artifacts` function to load the artifact. Always call load_artifacts
   before answering questions related to the artifacts, regardless of whether the
   artifacts have been loaded before. Do not depend on prior answers about the
   artifacts.
-  """])
+  """
+    llm_request._append_dynamic_instructions([instruction_text])
 
     # Attach the content of the artifacts if the model requests them.
     # This only adds the content to the model request, instead of the session.

src/google/adk/tools/load_mcp_resource_tool.py

@@ -110,13 +110,15 @@ async def _append_resources_to_llm_request(
     try:
       resource_names = await self._mcp_toolset.list_resources()
       if resource_names:
-        llm_request.append_instructions([f"""You have a list of MCP resources:
+        llm_request._append_dynamic_instructions(
+            [f"""You have a list of MCP resources:
 {json.dumps(resource_names)}
 
 When the user asks questions about any of the resources, you should call the
 `load_mcp_resource` function to load the resource. Always call load_mcp_resource
 before answering questions related to the resources.
-"""])
+"""]
+        )
     except Exception as e:
       logger.warning("Failed to list MCP resources: %s", e)
 

src/google/adk/tools/preload_memory_tool.py

@@ -86,7 +86,7 @@ async def process_llm_request(
 {full_memory_text}
 </PAST_CONVERSATIONS>
 """
-    llm_request.append_instructions([si])
+    llm_request._append_dynamic_instructions([si])
 
 
 preload_memory_tool = PreloadMemoryTool()

tests/unittests/flows/llm_flows/test_base_llm_flow.py

@@ -18,11 +18,15 @@
 from unittest import mock
 from unittest.mock import AsyncMock
 
+from google.adk.agents.invocation_context import InvocationContext
 from google.adk.agents.live_request_queue import LiveRequestQueue
 from google.adk.agents.llm_agent import Agent
 from google.adk.agents.loop_agent import LoopAgent
 from google.adk.agents.run_config import RunConfig
 from google.adk.events.event import Event
+from google.adk.features import FeatureName
+from google.adk.features._feature_registry import temporary_feature_override
+from google.adk.flows.llm_flows.base_llm_flow import _finalize_dynamic_instructions
 from google.adk.flows.llm_flows.base_llm_flow import _handle_after_model_callback
 from google.adk.flows.llm_flows.base_llm_flow import _process_agent_tools
 from google.adk.flows.llm_flows.base_llm_flow import _ReconnectSentinel
@@ -1922,3 +1926,90 @@ async def test_send_to_model_rejects_function_call():
       ValueError, match='User message cannot contain function calls'
   ):
     await flow._send_to_model(mock_connection, invocation_context)
+
+
+@pytest.mark.asyncio
+async def test_finalize_dynamic_instructions_feature_disabled():
+  """When feature flag is disabled, dynamic instructions append to system instruction."""
+
+  agent = Agent(name='test_agent', model='gemini-2.0-flash')
+
+  invocation_context = mock.Mock(spec=InvocationContext)
+  invocation_context.agent = agent
+
+  llm_request = LlmRequest()
+  llm_request._append_dynamic_instructions(['dynamic 1', 'dynamic 2'])
+  llm_request.contents.append(
+      types.Content(
+          role='user', parts=[types.Part.from_text(text='user question')]
+      )
+  )
+  with temporary_feature_override(
+      FeatureName.DYNAMIC_INSTRUCTION_ROUTING, False
+  ):
+    await _finalize_dynamic_instructions(invocation_context, llm_request)
+
+  assert llm_request.config.system_instruction is not None
+  assert len(llm_request.contents) == 1
+  assert llm_request.contents[0].parts[0].text == 'user question'
+
+
+@pytest.mark.asyncio
+async def test_finalize_dynamic_instructions_feature_enabled():
+  """When feature flag is enabled, dynamic instructions inject into contents."""
+
+  agent = Agent(name='test_agent', model='gemini-2.0-flash')
+
+  invocation_context = mock.Mock(spec=InvocationContext)
+  invocation_context.agent = agent
+
+  llm_request = LlmRequest()
+  llm_request._append_dynamic_instructions(['dynamic 1', 'dynamic 2'])
+  llm_request.contents.append(
+      types.Content(
+          role='user', parts=[types.Part.from_text(text='user question')]
+      )
+  )
+
+  with temporary_feature_override(
+      FeatureName.DYNAMIC_INSTRUCTION_ROUTING, True
+  ):
+    await _finalize_dynamic_instructions(invocation_context, llm_request)
+
+  assert llm_request.config.system_instruction is None
+  assert len(llm_request.contents) == 2
+  assert llm_request.contents[0].role == 'user'
+  assert llm_request.contents[0].parts[0].text == 'dynamic 1\n\ndynamic 2'
+  assert llm_request.contents[1].role == 'user'
+  assert llm_request.contents[1].parts[0].text == 'user question'
+
+
+@pytest.mark.asyncio
+async def test_finalize_dynamic_instructions_with_static_instruction():
+  """When static_instruction is set and feature flag enabled, it injects into contents."""
+
+  agent = Agent(name='test_agent', model='gemini-2.0-flash')
+  agent.static_instruction = 'static content'
+
+  invocation_context = mock.Mock(spec=InvocationContext)
+  invocation_context.agent = agent
+
+  llm_request = LlmRequest()
+  llm_request._append_dynamic_instructions(['dynamic 1', 'dynamic 2'])
+  llm_request.contents.append(
+      types.Content(
+          role='user', parts=[types.Part.from_text(text='user question')]
+      )
+  )
+
+  with temporary_feature_override(
+      FeatureName.DYNAMIC_INSTRUCTION_ROUTING, True
+  ):
+    await _finalize_dynamic_instructions(invocation_context, llm_request)
+
+  assert llm_request.config.system_instruction is None
+  assert len(llm_request.contents) == 2
+  assert llm_request.contents[0].role == 'user'
+  assert llm_request.contents[0].parts[0].text == 'dynamic 1\n\ndynamic 2'
+  assert llm_request.contents[1].role == 'user'
+  assert llm_request.contents[1].parts[0].text == 'user question'

tests/unittests/tools/test_load_artifacts_tool.py

@@ -453,3 +453,20 @@ def test_get_declaration_with_json_schema_feature_enabled():
           },
       },
   }
+
+
+@mark.asyncio
+async def test_load_artifacts_registers_dynamic_instructions():
+  """load_artifacts registers instructions in llm_request._dynamic_instructions."""
+  tool_context = _StubToolContext(
+      {'doc.txt': types.Part.from_text(text='hello')},
+  )
+  llm_request = LlmRequest()
+  await load_artifacts_tool.process_llm_request(
+      tool_context=tool_context, llm_request=llm_request
+  )
+
+  assert len(llm_request._dynamic_instructions) == 1
+  assert 'You have a list of artifacts' in llm_request._dynamic_instructions[0]
+  assert llm_request.config.system_instruction is None
+  assert len(llm_request.contents) == 0

tests/unittests/tools/test_load_mcp_resource_tool.py

@@ -80,8 +80,8 @@ async def test_process_llm_request_injects_list(self):
         tool_context=self.mock_tool_context, llm_request=llm_request
     )
 
-    llm_request.append_instructions.assert_called_once()
-    instructions = llm_request.append_instructions.call_args[0][0]
+    llm_request._append_dynamic_instructions.assert_called_once()
+    instructions = llm_request._append_dynamic_instructions.call_args[0][0]
     assert "res1" in instructions[0]
     assert "res2" in instructions[0]
 

tests/unittests/tools/test_load_memory_tool.py

@@ -12,10 +12,17 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from unittest import mock
+from unittest.mock import AsyncMock
+
 from google.adk.features import FeatureName
 from google.adk.features._feature_registry import temporary_feature_override
+from google.adk.models.llm_request import LlmRequest
 from google.adk.tools.load_memory_tool import load_memory_tool
+from google.adk.tools.preload_memory_tool import PreloadMemoryTool
+from google.adk.tools.tool_context import ToolContext
 from google.genai import types
+import pytest
 
 
 def test_get_declaration_with_json_schema_feature_disabled():
@@ -44,3 +51,36 @@ def test_get_declaration_with_json_schema_feature_enabled():
       },
       'required': ['query'],
   }
+
+
+@pytest.mark.asyncio
+async def test_preload_memory_registers_dynamic_instructions():
+  """Test that PreloadMemoryTool registers memory into _dynamic_instructions."""
+  tool = PreloadMemoryTool()
+  tool_context = mock.Mock(spec=ToolContext)
+  tool_context.user_content = types.Content(
+      role='user', parts=[types.Part.from_text(text='my query')]
+  )
+  mock_memory = mock.Mock()
+  mock_memory.timestamp = None
+  mock_memory.author = 'user'
+  mock_memory.content = types.Content(
+      parts=[
+          types.Part.from_text(text='User prefers Python programming language.')
+      ]
+  )
+
+  tool_context.search_memory = AsyncMock(
+      return_value=mock.Mock(memories=[mock_memory])
+  )
+
+  llm_request = LlmRequest()
+
+  await tool.process_llm_request(
+      tool_context=tool_context, llm_request=llm_request
+  )
+
+  assert len(llm_request._dynamic_instructions) == 1
+  assert '<PAST_CONVERSATIONS>' in llm_request._dynamic_instructions[0]
+  assert llm_request.config.system_instruction is None
+  assert len(llm_request.contents) == 0