feat(agents): add include_contents='current' for invocation-scoped context

Add 'current' as a third value for include_contents alongside 'default'
and 'none'. Where 'none' anchors at the last turn boundary (user OR
other-agent event), 'current' anchors at the last user message, giving
the agent everything from the user's request through all sibling agent
outputs within the same invocation.

This fixes the composition footgun with include_sources: 'none' +
include_sources=['user'] returns empty context when the last event is a
filtered other-agent output. With 'current', the boundary is always a
user event so filtering never produces an empty context.

A UserWarning is raised at construction time when include_contents='none'
is combined with include_sources, since the turn boundary may land on a
filtered event and produce empty context at runtime.

Author: bofenghuang

src/google/adk/agents/llm_agent.py

@@ -343,13 +343,15 @@ class LlmAgent(BaseAgent, abc.ABC):
   """Disallows LLM-controlled transferring to the peer agents."""
   # LLM-based agent transfer configs - End
 
-  include_contents: Literal['default', 'none'] = 'default'
+  include_contents: Literal['default', 'current', 'none'] = 'default'
   """Controls content inclusion in model requests.
 
   Options:
-    default: Model receives relevant conversation history
-    none: Model receives no prior history, operates solely on current
-    instruction and input
+    default: Model receives full conversation history.
+    current: Model receives all events since the last user message,
+      including outputs from agents that ran earlier in this pipeline.
+    none: Model receives only the most recent agent or user input, with no
+      prior conversation history.
   """
 
   include_sources: Optional[list[str]] = None
@@ -976,6 +978,16 @@ def __maybe_save_output_to_state(self, event: Event):
 
   @model_validator(mode='after')
   def __model_validator_after(self) -> LlmAgent:
+    if self.include_contents == 'none' and self.include_sources is not None:
+      warnings.warn(
+          "include_contents='none' with include_sources may produce empty"
+          ' context: the turn boundary is the last user OR other-agent event,'
+          ' and if that event is filtered by include_sources the context will'
+          " be empty. Use include_contents='current' to anchor at the last"
+          ' user message instead.',
+          UserWarning,
+          stacklevel=2,
+      )
     return self
 
   @field_validator('include_sources', mode='after')  # type: ignore[misc]

src/google/adk/agents/llm_agent_config.py

@@ -121,7 +121,7 @@ def _validate_model_sources(self) -> LlmAgentConfig:
       default=None, description='Optional. LlmAgent.output_key.'
   )
 
-  include_contents: Literal['default', 'none'] = Field(
+  include_contents: Literal['default', 'current', 'none'] = Field(
       default='default', description='Optional. LlmAgent.include_contents.'
   )
 

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

@@ -82,7 +82,10 @@ async def run_async(
           source_filter=source_filter,
       )
     else:
-      # Include current turn context only (no conversation history)
+      # 'current': anchor at last user message — all sibling agent outputs
+      #            within this invocation are included.
+      # 'none':    anchor at last turn boundary (user OR other-agent event).
+      stop_at_user_only = agent.include_contents == 'current'
       llm_request.contents = _get_current_turn_contents(
           invocation_context.branch,
           invocation_context.session.events,
@@ -92,6 +95,7 @@ async def run_async(
           is_single_turn=is_single_turn,
           user_content=invocation_context.user_content,
           source_filter=source_filter,
+          stop_at_user_only=stop_at_user_only,
       )
 
     # Add instruction-related contents to proper position in conversation
@@ -718,33 +722,41 @@ def _get_current_turn_contents(
     isolation_scope: Optional[str] = None,
     user_content: Optional[types.Content] = None,
     source_filter: Optional[list[str]] = None,
+    stop_at_user_only: bool = False,
 ) -> list[types.Content]:
   """Get contents for the current turn only (no conversation history).
 
-  When include_contents='none', we want to include:
-  - The current user input
-  - Tool calls and responses from the current turn
-  But exclude conversation history from previous turns.
-
-  In multi-agent scenarios, the "current turn" for an agent starts from an
-  actual user or from another agent.
+  Used by include_contents='none' and 'current'. Both exclude prior-session
+  history; they differ in the turn boundary:
+    'none'    (stop_at_user_only=False): last user OR other-agent event.
+    'current' (stop_at_user_only=True):  last user event only.
 
   Args:
     current_branch: The current branch of the agent.
     events: A list of all session events.
     agent_name: The name of the agent.
     preserve_function_call_ids: Whether to preserve function call ids.
+    stop_at_user_only: When True, anchor only at user events ('current' mode).
 
   Returns:
-    A list of contents for the current turn only, preserving context needed
-    for proper tool execution while excluding conversation history.
+    A list of contents from the turn boundary forward. Returns [] if no
+    qualifying boundary event is found.
   """
-  # Find the latest event that starts the current turn and process from there
+  # Find the latest event that starts the current turn and process from there.
+  # stop_at_user_only=True ('current' mode): anchor at last user message,
+  # so all sibling agent outputs within this invocation are included.
+  # stop_at_user_only=False ('none' mode): anchor at last user OR other-agent.
   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)):
+    is_turn_start = event.author == 'user' or (
+        not stop_at_user_only and _is_other_agent_reply(agent_name, event)
+    )
+    if (
+        _should_include_event_in_context(
+            current_branch, event, isolation_scope=isolation_scope
+        )
+        and is_turn_start
+    ):
       return _get_contents(
           current_branch,
           events[i:],

tests/unittests/agents/test_llm_agent_include_contents.py

@@ -355,3 +355,195 @@ async def test_include_sources_user_self_drops_upstream_across_turns():
       "upstream reply" in str(c).lower() for _, c in agent2_second_contents
   )
   assert not any("For context:" in str(c) for _, c in agent2_second_contents)
+
+
+# ---------------------------------------------------------------------------
+# include_contents='default' + include_sources combinations
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_include_contents_default_with_source_filter_user():
+  """include_contents='default' + include_sources=['user'] keeps only user messages across full history."""
+  agent1_model = testing_utils.MockModel.create(
+      responses=["Agent1 result turn1", "Agent1 result turn2"]
+  )
+  agent1 = LlmAgent(
+      name="agent1", model=agent1_model, instruction="You are agent1"
+  )
+
+  agent2_model = testing_utils.MockModel.create(
+      responses=["Turn1 response", "Turn2 response"]
+  )
+  agent2 = LlmAgent(
+      name="agent2",
+      model=agent2_model,
+      include_sources=["user"],
+      instruction="You are agent2",
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      SequentialAgent(name="pipeline", sub_agents=[agent1, agent2])
+  )
+  runner.run("First user message")
+  runner.run("Second user message")
+
+  # Second invocation: full history, but only user messages kept
+  agent2_second_contents = testing_utils.simplify_contents(
+      agent2_model.requests[1].contents
+  )
+  assert any("First user message" in str(c) for _, c in agent2_second_contents)
+  assert any("Second user message" in str(c) for _, c in agent2_second_contents)
+  assert not any("Agent1 result" in str(c) for _, c in agent2_second_contents)
+  assert not any("For context:" in str(c) for _, c in agent2_second_contents)
+
+
+# ---------------------------------------------------------------------------
+# include_contents='current'
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_include_contents_current_sees_user_and_all_upstream_agents():
+  """include_contents='current' anchors at user message — all sibling agents visible."""
+  agent1_model = testing_utils.MockModel.create(responses=["Agent1 result"])
+  agent1 = LlmAgent(
+      name="agent1", model=agent1_model, instruction="You are agent1"
+  )
+
+  agent2_model = testing_utils.MockModel.create(responses=["Agent2 result"])
+  agent2 = LlmAgent(
+      name="agent2", model=agent2_model, instruction="You are agent2"
+  )
+
+  agent3_model = testing_utils.MockModel.create(responses=["Agent3 done"])
+  agent3 = LlmAgent(
+      name="agent3",
+      model=agent3_model,
+      include_contents="current",
+      instruction="You are agent3",
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      SequentialAgent(name="pipeline", sub_agents=[agent1, agent2, agent3])
+  )
+  runner.run("Original user request")
+
+  agent3_contents = testing_utils.simplify_contents(
+      agent3_model.requests[0].contents
+  )
+
+  # User message must be present
+  assert any("Original user request" in str(c) for _, c in agent3_contents)
+  # Both upstream agents' narrative entries must be present
+  assert any("Agent1 result" in str(c) for _, c in agent3_contents)
+  assert any("Agent2 result" in str(c) for _, c in agent3_contents)
+
+
+@pytest.mark.asyncio
+async def test_include_contents_current_with_source_filter_user_not_empty():
+  """include_contents='current' + include_sources=['user'] → user message, not empty.
+
+  Contrast with include_contents='none' + include_sources=['user'] which
+  produces empty context when the last event is a peer agent's output.
+  """
+  agent1_model = testing_utils.MockModel.create(responses=["Agent1 result"])
+  agent1 = LlmAgent(
+      name="agent1", model=agent1_model, instruction="You are agent1"
+  )
+
+  agent2_model = testing_utils.MockModel.create(responses=["Agent2 done"])
+  agent2 = LlmAgent(
+      name="agent2",
+      model=agent2_model,
+      include_contents="current",
+      include_sources=["user"],
+      instruction="You are agent2",
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      SequentialAgent(name="pipeline", sub_agents=[agent1, agent2])
+  )
+  runner.run("Hello from user")
+
+  agent2_contents = testing_utils.simplify_contents(
+      agent2_model.requests[0].contents
+  )
+
+  # User message must be present and result must not be empty
+  assert len(agent2_contents) > 0
+  assert any("Hello from user" in str(c) for _, c in agent2_contents)
+  # Upstream agent narrative must be filtered out
+  assert not any("Agent1 result" in str(c) for _, c in agent2_contents)
+  assert not any("For context:" in str(c) for _, c in agent2_contents)
+
+
+@pytest.mark.asyncio
+async def test_include_contents_current_with_source_filter_user_and_self():
+  """include_contents='current' + include_sources=['user', 'self'] keeps user + own turns only."""
+  agent1_model = testing_utils.MockModel.create(
+      responses=["Agent1 result turn1", "Agent1 result turn2"]
+  )
+  agent1 = LlmAgent(
+      name="agent1", model=agent1_model, instruction="You are agent1"
+  )
+
+  agent2_model = testing_utils.MockModel.create(
+      responses=["Agent2 first turn", "Agent2 second turn"]
+  )
+  agent2 = LlmAgent(
+      name="agent2",
+      model=agent2_model,
+      include_contents="current",
+      include_sources=["user", "self"],
+      instruction="You are agent2",
+  )
+
+  runner = testing_utils.InMemoryRunner(
+      SequentialAgent(name="pipeline", sub_agents=[agent1, agent2])
+  )
+  runner.run("First user message")
+  runner.run("Second user message")
+
+  # Second invocation: current window starts at 'Second user message'.
+  # Agent2's own prior turn from invocation 1 is outside that window — naturally absent.
+  # What we verify: user present, upstream agent filtered by include_sources.
+  agent2_second_contents = testing_utils.simplify_contents(
+      agent2_model.requests[1].contents
+  )
+  assert any("Second user message" in str(c) for _, c in agent2_second_contents)
+  assert not any("Agent1 result" in str(c) for _, c in agent2_second_contents)
+  assert not any("For context:" in str(c) for _, c in agent2_second_contents)
+
+
+def test_include_contents_none_with_include_sources_warns():
+  """include_contents='none' + include_sources triggers a UserWarning."""
+  import warnings as _warnings
+
+  with _warnings.catch_warnings(record=True) as w:
+    _warnings.simplefilter("always")
+    LlmAgent(
+        name="agent",
+        model="gemini-2.5-flash",
+        include_contents="none",
+        include_sources=["user"],
+    )
+  assert len(w) == 1
+  assert issubclass(w[0].category, UserWarning)
+  assert "include_contents='current'" in str(w[0].message)
+
+
+def test_include_contents_none_with_agent_name_in_sources_still_warns():
+  """Warning fires even with a concrete agent name — still risky at runtime."""
+  import warnings as _warnings
+
+  with _warnings.catch_warnings(record=True) as w:
+    _warnings.simplefilter("always")
+    LlmAgent(
+        name="agent",
+        model="gemini-2.5-flash",
+        include_contents="none",
+        include_sources=["user", "upstream_agent"],
+    )
+  assert len(w) == 1
+  assert issubclass(w[0].category, UserWarning)