fix: Retain user input across transfer_to_agent when include_contents='none'

When an LlmAgent is configured with include_contents='none', the documented
behavior is that the model receives no prior history and operates only on the
current instruction and the latest user input. However, during an agent handoff
via transfer_to_agent, the sub-agent's "current turn" incorrectly anchored on
the parent's trailing transfer_to_agent events (the function call and its
response) instead of the latest user input, so the user input was dropped from
the sub-agent's context.

Fix _get_current_turn_contents to skip trailing direct transfer_to_agent events
when locating the start of the current turn, so the turn anchors on the latest
user input (or a non-transfer model request). The transfer events are still
included as context.

Closes #3535

Co-authored-by: Haran Rajkumar <haranrk@google.com>
PiperOrigin-RevId: 944016382

Author: 2 people

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

@@ -776,9 +776,13 @@ def _get_current_turn_contents(
   # Find the latest event that starts the current turn and process from there
   for i in range(len(events) - 1, -1, -1):
     event = events[i]
-    if _should_include_event_in_context(
-        current_branch, event, isolation_scope=isolation_scope
-    ) and (event.author == 'user' or _is_other_agent_reply(agent_name, event)):
+    if (
+        _should_include_event_in_context(
+            current_branch, event, isolation_scope=isolation_scope
+        )
+        and (event.author == 'user' or _is_other_agent_reply(agent_name, event))
+        and not _is_direct_transfer(event)
+    ):
       return _get_contents(
           current_branch,
           events[i:],
@@ -792,6 +796,30 @@ def _get_current_turn_contents(
   return []
 
 
+def _is_direct_transfer(event: Event) -> bool:
+  """Whether the event is a direct ``transfer_to_agent`` event.
+
+  When ``include_contents='none'`` and control is handed to a sub-agent via
+  ``transfer_to_agent``, the trailing transfer events (the function call and
+  its response) must not be treated as the start of the current turn.
+  Otherwise the sub-agent's turn would anchor on the parent's transfer event
+  and drop the latest user input. Skipping these events lets the turn anchor
+  on the real user input (or a non-transfer model request) instead, while the
+  transfer events are still included as context.
+  """
+  return bool(
+      event.actions.transfer_to_agent
+      or (
+          event.content
+          and event.content.parts
+          and any(
+              p.function_call and p.function_call.name == 'transfer_to_agent'
+              for p in event.content.parts
+          )
+      )
+  )
+
+
 def _is_other_agent_reply(current_agent_name: str, event: Event) -> bool:
   """Whether the event is a reply from another agent."""
   # In live/bidi mode, all events from any agents, including the current

tests/unittests/flows/llm_flows/test_contents.py

@@ -254,6 +254,99 @@ async def test_include_contents_none_multi_branch_current_turn():
   ]
 
 
+@pytest.mark.asyncio
+async def test_events_with_transfer_to_agent_are_included():
+  """Test that the user input is retained across a transfer_to_agent handoff.
+
+  When include_contents='none' and control is transferred to a sub-agent, the
+  current turn must anchor on the latest user input rather than the trailing
+  transfer_to_agent events, while still including those transfer events as
+  context.
+  """
+  agent = Agent(
+      model="gemini-2.5-flash", name="test_agent", include_contents="none"
+  )
+  llm_request = LlmRequest(model="gemini-2.5-flash")
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent
+  )
+
+  events = [
+      Event(
+          invocation_id="inv1",
+          author="user",
+          content=types.UserContent("First user message"),
+      ),
+      Event(
+          invocation_id="inv1",
+          author="parent",
+          content=types.Content(
+              parts=[
+                  types.Part(
+                      function_call=types.FunctionCall(
+                          args={"agent_name": "test_agent"},
+                          id="call_inv1",
+                          name="transfer_to_agent",
+                      )
+                  )
+              ],
+              role="model",
+          ),
+      ),
+      Event(
+          invocation_id="inv1",
+          author="parent",
+          content=types.Content(
+              parts=[
+                  types.Part(
+                      function_response=types.FunctionResponse(
+                          id="call_inv1",
+                          name="transfer_to_agent",
+                          response={"result": None},
+                      ),
+                  ),
+              ],
+              role="user",
+          ),
+          actions=EventActions(transfer_to_agent="test_agent"),
+      ),
+  ]
+
+  invocation_context.session.events = events
+  async for _ in contents.request_processor.run_async(
+      invocation_context, llm_request
+  ):
+    pass
+
+  assert llm_request.contents == [
+      types.UserContent("First user message"),
+      types.Content(
+          parts=[
+              types.Part(text="For context:"),
+              types.Part(
+                  text=(
+                      "[parent] called tool `transfer_to_agent` with"
+                      " parameters: {'agent_name': 'test_agent'}"
+                  )
+              ),
+          ],
+          role="user",
+      ),
+      types.Content(
+          parts=[
+              types.Part(text="For context:"),
+              types.Part(
+                  text=(
+                      "[parent] `transfer_to_agent` tool returned result:"
+                      " {'result': None}"
+                  )
+              ),
+          ],
+          role="user",
+      ),
+  ]
+
+
 @pytest.mark.asyncio
 async def test_authentication_events_are_filtered():
   """Test that authentication function calls and responses are filtered out."""