fix(a2a) warn when session state is dropped

docs/guides/agents/remote_a2a_agent/task.md

@@ -196,6 +196,21 @@ behavior:
 
 ## Limitations
 
+### Session state boundary
+
+The caller and remote agent use separate sessions. Session state and
+`EventActions.state_delta` do not cross the A2A boundary in either direction:
+
+- `output_key` on the remote agent writes only to the remote server's session.
+- A caller-side state-only event has no content to include in the A2A request.
+- A state delta supplied by a remote peer is not applied to the caller's
+  session, because peers are not allowed to mutate caller state.
+
+Pass values required by the remote agent in event content. Return values needed
+by the caller as response content or, in task mode, as `finish_task` output.
+ADK logs a warning when it is about to drop a state-only hand-off or receives a
+remote state delta.
+
 -   **Workflow Graphs Not Supported**: `RemoteA2aAgent` in task mode
     (`mode="task"`) cannot be used as a node in ADK `Workflow` graphs. It is
     exclusively designed for sub-agent delegation under a parent coordinator

src/google/adk/a2a/converters/to_adk_event.py

@@ -321,10 +321,21 @@ def _extract_event_actions(metadata: Any) -> EventActions:
       for key, value in parsed_actions.items()
       if key in _PEER_SETTABLE_ACTION_FIELDS
   }
-  if len(peer_actions) != len(parsed_actions):
+  dropped_fields = set(parsed_actions) - set(peer_actions)
+  state_delta_alias = EventActions.model_fields["state_delta"].alias
+  state_delta_fields = {"state_delta"}
+  if state_delta_alias:
+    state_delta_fields.add(state_delta_alias)
+  if dropped_fields & state_delta_fields:
+    logger.warning(
+        "Ignoring a session state delta from a remote A2A peer. Session state"
+        " is local to each agent; return values needed by the caller as event"
+        " content or task output instead."
+    )
+  if dropped_fields:
     logger.debug(
         "Dropping ADK actions metadata fields that a peer may not set: %s",
-        sorted(set(parsed_actions) - set(peer_actions)),
+        sorted(dropped_fields),
     )
 
   try:

src/google/adk/agents/remote_a2a_agent.py

@@ -563,6 +563,11 @@ async def _before_request(
 class RemoteA2aAgent(BaseAgent):
   """Agent that communicates with a remote A2A agent via A2A client.
 
+  Session state is local to each side of an A2A boundary. Only event content is
+  included in requests, and state deltas from a remote peer are not applied to
+  the caller's session. Put values needed by the peer in event content and
+  return values needed by the caller as content or task output.
+
   This agent supports multiple ways to specify the remote agent:
   1. Direct AgentCard object
   2. URL to agent card JSON
@@ -571,7 +576,6 @@ class RemoteA2aAgent(BaseAgent):
   The agent handles:
   - Agent card resolution and validation
   - HTTP client management with proper resource cleanup
-  - A2A message conversion and error handling
   - Session state management across requests
   """
 
@@ -1172,6 +1176,18 @@ def _construct_message_parts_from_session(
           " session history. Workflow path scopes are not supported."
       )
 
+    if events_to_process:
+      last_event = events_to_process[0]
+      has_state_delta = bool(last_event.actions.state_delta)
+      has_content = bool(last_event.content and last_event.content.parts)
+      if has_state_delta and not has_content:
+        logger.warning(
+            "RemoteA2aAgent '%s' cannot forward the preceding state-only"
+            " event across A2A. Session state is local to each agent; include"
+            " the required values in event content instead.",
+            self.name,
+        )
+
     # Collect all FC IDs emitted by this remote agent in the task scope.
     remote_fc_ids = set()
     if self.mode == "task":

tests/unittests/a2a/converters/test_to_adk.py

@@ -430,6 +430,33 @@ def test_peer_supplied_actions_cannot_mutate_caller_session(self):
       # Inert fields a peer may set are still honored.
       assert event.actions.escalate is True
 
+  def test_peer_state_delta_logs_session_boundary_warning(self, caplog):
+    metadata = {
+        _get_adk_metadata_key("actions"): {
+            "stateDelta": {"remote_result": "value"}
+        }
+    }
+    message = Message(
+        message_id="msg-1",
+        role=_compat.ROLE_AGENT,
+        parts=[_make_a2a_part_for_test({})],
+        metadata=metadata,
+    )
+
+    with caplog.at_level("WARNING"):
+      event = convert_a2a_message_to_event(
+          message,
+          "test-author",
+          self.mock_context,
+          Mock(return_value=[genai_types.Part.from_text(text="result")]),
+      )
+
+    assert event is not None
+    assert event.actions.state_delta == {}
+    assert (
+        "Ignoring a session state delta from a remote A2A peer" in caplog.text
+    )
+
   def test_peer_settable_action_fields_are_exactly_inert(self):
     """Test the peer allow-list holds every spelling of the inert fields."""
     inert_fields = {"escalate", "skip_summarization"}

tests/unittests/agents/test_remote_a2a_agent.py

@@ -29,7 +29,6 @@
 from a2a.types import AgentCard
 from a2a.types import AgentInterface
 from a2a.types import AgentSkill
-from a2a.types import Artifact
 from a2a.types import Message as A2AMessage
 from a2a.types import Task as A2ATask
 from a2a.types import TaskArtifactUpdateEvent
@@ -59,6 +58,7 @@
 from google.adk.auth.auth_credential import OAuth2Auth
 from google.adk.auth.auth_preprocessor import TOOLSET_AUTH_CREDENTIAL_ID_PREFIX
 from google.adk.events.event import Event
+from google.adk.events.event_actions import EventActions
 from google.adk.flows.llm_flows.functions import REQUEST_EUC_FUNCTION_CALL_NAME
 from google.adk.sessions.session import Session
 from google.genai import types as genai_types
@@ -1298,6 +1298,24 @@ def test_construct_message_parts_from_session_empty_events(self):
     assert parts == []
     assert context_id is None
 
+  def test_construct_message_parts_warns_for_state_only_handoff(self, caplog):
+    """A state-only hand-off warns before the remote receives stale content."""
+    self.mock_session.events = [
+        Event(
+            author="local_agent",
+            actions=EventActions(state_delta={"routing": "priority"}),
+        )
+    ]
+
+    with caplog.at_level("WARNING"):
+      parts, context_id = self.agent._construct_message_parts_from_session(
+          self.mock_context
+      )
+
+    assert parts == []
+    assert context_id is None
+    assert "cannot forward the preceding state-only event" in caplog.text
+
   def test_construct_message_parts_from_session_foreign_function_response_not_converted(
       self,
   ):

tests/unittests/agents/test_remote_a2a_agent_state_boundary.py

@@ -0,0 +1,69 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from unittest.mock import Mock
+
+from google.adk.a2a import _compat
+from google.adk.agents.invocation_context import InvocationContext
+from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
+from google.adk.events.event import Event
+from google.adk.events.event_actions import EventActions
+from google.adk.sessions.session import Session
+from google.genai import types as genai_types
+
+
+def test_task_mode_state_warning_uses_latest_applicable_event(caplog):
+  task_scope = "task-scope"
+  agent = RemoteA2aAgent(
+      name="remote",
+      agent_card="https://example.com/.well-known/agent-card.json",
+      genai_part_converter=lambda _: _compat.make_text_part("converted"),
+  )
+  agent.mode = "task"
+
+  trigger = Event(
+      author="coordinator",
+      content=genai_types.Content(
+          parts=[
+              genai_types.Part(
+                  function_call=genai_types.FunctionCall(
+                      id=task_scope,
+                      name=agent.name,
+                      args={},
+                  )
+              )
+          ]
+      ),
+  )
+  applicable = Event(
+      author="user",
+      isolation_scope=task_scope,
+      content=genai_types.Content(parts=[genai_types.Part(text="hello")]),
+  )
+  unrelated_state_only = Event(
+      author="other",
+      isolation_scope="different-task",
+      actions=EventActions(state_delta={"routing": "priority"}),
+  )
+
+  session = Mock(spec=Session)
+  session.events = [trigger, applicable, unrelated_state_only]
+  ctx = Mock(spec=InvocationContext)
+  ctx.session = session
+  ctx.isolation_scope = task_scope
+
+  with caplog.at_level("WARNING"):
+    agent._construct_message_parts_from_session(ctx)
+
+  assert "cannot forward the preceding state-only event" not in caplog.text