fix(workflow): stop run_llm_agent_as_node mutating the shared node agent

src/google/adk/workflow/_llm_agent_wrapper.py

@@ -410,8 +410,6 @@ async def run_llm_agent_as_node(
     )
 
   include_contents_explicit = 'include_contents' in agent.model_fields_set
-  if agent.mode == 'single_turn' and not include_contents_explicit:
-    agent.include_contents = 'none'
 
   agent_ctx = prepare_llm_agent_context(agent, ctx)
   prepare_llm_agent_input(agent, agent_ctx, node_input)
@@ -446,9 +444,26 @@ async def run_llm_agent_as_node(
 
   if agent.mode == 'single_turn':
     # is_live is always False here (single_turn forces non-live).
-    async with aclosing(agent.run_async(ic)) as run_iter:
+    #
+    # A node in single_turn mode with no explicit include_contents defaults
+    # to 'none'. That default is invocation-scoped, not part of the agent's
+    # own configuration, so it is applied on a per-invocation clone rather
+    # than by mutating `agent` itself: `agent` is the same node instance
+    # reused across every future run of this workflow, and mutating it here
+    # would leave the override permanently in place for every other
+    # invocation of that shared node, single_turn or not.
+    if include_contents_explicit:
+      run_agent = agent
+    else:
+      run_agent = agent.clone(update={'include_contents': 'none'})
+      # clone() drops parent_agent (it assumes the caller is defining a new,
+      # independent agent); this is a same-invocation stand-in for `agent`,
+      # so it must keep the same parent as the original. See build_node's
+      # identical restoration for the same reason.
+      run_agent.parent_agent = agent.parent_agent
+    async with aclosing(run_agent.run_async(ic)) as run_iter:
       async for event in run_iter:
-        process_llm_agent_output(agent, ctx, event)
+        process_llm_agent_output(run_agent, ctx, event)
         yield event
     return
 

tests/unittests/workflow/test_llm_agent_as_node.py

@@ -22,6 +22,7 @@
 from __future__ import annotations
 
 from typing import Any
+from unittest.mock import MagicMock
 
 from google.adk.agents.context import Context
 from google.adk.agents.llm.task._task_models import TaskResult
@@ -320,27 +321,32 @@ async def test_single_turn_defaults_include_contents_only_when_unset(
       agent_kwargs: dict[str, Any],
       expected_include_contents: str,
   ):
-    """Single-turn workflow nodes preserve explicit content inclusion."""
-    from unittest.mock import MagicMock
+    """Single-turn nodes get the right effective include_contents per run,
 
+    without permanently mutating the shared node object itself: a node is
+    reused across every future invocation, so an implicit 'none' default
+    applied by mutating it in place would stick for every later run,
+    single_turn or not, explicit or not.
+    """
     agent = LlmAgent(
         name='test_agent',
         model='gemini-2.5-flash',
         instruction='Test.',
         **agent_kwargs,
     )
     wrapper = build_node(agent)
+    original_include_contents = wrapper.include_contents
     seen_include_contents = []
 
-    async def mock_run_async(*args, **kwargs):
-      seen_include_contents.append(wrapper.include_contents)
+    async def mock_run_async(self, *args, **kwargs):
+      seen_include_contents.append(self.include_contents)
       yield Event(
           invocation_id='inv',
-          author=wrapper.name,
+          author=self.name,
           content=types.Content(parts=[types.Part(text='ok')]),
       )
 
-    object.__setattr__(wrapper, 'run_async', mock_run_async)
+    monkeypatch.setattr(LlmAgent, 'run_async', mock_run_async)
     monkeypatch.setattr(
         agent_wrapper,
         'prepare_llm_agent_context',
@@ -360,8 +366,11 @@ async def mock_run_async(*args, **kwargs):
         event async for event in wrapper._run_impl(ctx=ctx, node_input='hi')
     ]
 
+    # The effective value used for this run is still correct...
     assert seen_include_contents == [expected_include_contents]
-    assert wrapper.include_contents == expected_include_contents
+    # ...but the shared node itself is never mutated, regardless of whether
+    # this run needed an implicit override.
+    assert wrapper.include_contents == original_include_contents
     assert events[0].content.parts[0].text == 'ok'
 
   def test_name_override(self):