fix: resume paused A2A tasks on ordinary user replies

src/google/adk/a2a/_compat.py

@@ -148,6 +148,7 @@ def _proto_to_dict(msg: Any) -> dict[str, Any]:
   TS_INPUT_REQUIRED = TaskState.Value("TASK_STATE_INPUT_REQUIRED")
   TS_AUTH_REQUIRED = TaskState.Value("TASK_STATE_AUTH_REQUIRED")
   TS_CANCELED = TaskState.Value("TASK_STATE_CANCELED")
+  TS_REJECTED = TaskState.Value("TASK_STATE_REJECTED")
 
   TP_JSONRPC = TransportProtocol.JSONRPC
   TP_HTTP_JSON = TransportProtocol.HTTP_JSON
@@ -164,6 +165,7 @@ def _proto_to_dict(msg: Any) -> dict[str, Any]:
   TS_INPUT_REQUIRED = TaskState.input_required
   TS_AUTH_REQUIRED = TaskState.auth_required
   TS_CANCELED = TaskState.canceled
+  TS_REJECTED = TaskState.rejected
 
   TP_JSONRPC = getattr(TransportProtocol, "jsonrpc")
   TP_HTTP_JSON = getattr(TransportProtocol, "http_json")

src/google/adk/a2a/agent/_remote_a2a_agent.py

@@ -33,6 +33,7 @@
 from a2a.types import AgentCard
 from a2a.types import Message as A2AMessage
 from a2a.types import Part as A2APart
+from a2a.types import Task as A2ATask
 from a2a.types import TaskArtifactUpdateEvent as A2ATaskArtifactUpdateEvent
 from a2a.types import TaskState
 from a2a.types import TaskStatusUpdateEvent as A2ATaskStatusUpdateEvent
@@ -77,6 +78,7 @@
 from ..converters.part_converter import convert_genai_part_to_a2a_part
 from ..converters.part_converter import GenAIPartToA2APartConverter
 from ..converters.to_adk_event import _create_mock_function_call_for_required_user_input
+from ..converters.to_adk_event import _TASK_RESPONSE_BOUNDARY_STATES
 from ..converters.to_adk_event import MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_AUTH
 from ..converters.to_adk_event import MOCK_FUNCTION_CALL_FOR_REQUIRED_USER_INPUT
 from ..experimental import a2a_experimental
@@ -147,6 +149,18 @@
     _compat.TS_WORKING,
 })
 
+
+def _task_response_metadata(task: A2ATask) -> dict[str, Any]:
+  """Records the latest task snapshot, including contentless status updates."""
+  metadata = {
+      A2A_METADATA_PREFIX + "task_id": task.id,
+      A2A_METADATA_PREFIX + "response": _compat.a2a_to_dict(task),
+  }
+  if task.context_id:
+    metadata[A2A_METADATA_PREFIX + "context_id"] = task.context_id
+  return metadata
+
+
 _RESULT_KEY = "result"
 
 # Top-level keys of a serialized AuthConfig, the shape an adk_request_credential
@@ -1381,6 +1395,73 @@ def _build_message_parts_for_request(
       return self._context_builder(ctx, self.name, self._genai_part_converter)
     return self._construct_message_parts_from_session(ctx)
 
+  def _get_resumable_task_id(
+      self, ctx: InvocationContext, context_id: Optional[str]
+  ) -> Optional[str]:
+    """Returns the latest paused task in the outgoing message's context."""
+    if not context_id or not self._has_default_response_persistence():
+      return None
+    task_scope = ctx.isolation_scope if self.mode == "task" else None
+    for event in reversed(ctx.session.events):
+      if task_scope and event.isolation_scope != task_scope:
+        # Do not resume a task from an earlier delegation to the same agent.
+        if any(fc.id == task_scope for fc in event.get_function_calls()):
+          return None
+        continue
+      metadata = event.custom_metadata or {}
+      if (
+          event.author == self.name
+          and event.error_message
+          and metadata.get(A2A_METADATA_PREFIX + "request")
+      ):
+        # A failed request may already have consumed the recorded pause.
+        # Its old task state is no longer sufficient to infer a continuation.
+        return None
+      if not self._is_remote_response(event):
+        continue
+      if metadata.get(A2A_METADATA_PREFIX + "context_id") != context_id:
+        return None
+      response = metadata.get(A2A_METADATA_PREFIX + "response")
+      status = response.get("status") if isinstance(response, dict) else None
+      # Session metadata contains the serialized wire state. A2A 0.3 uses
+      # hyphenated names, while A2A 1.x uses protobuf enum names.
+      if not isinstance(status, dict) or status.get("state") not in (
+          "input-required",
+          "auth-required",
+          "TASK_STATE_INPUT_REQUIRED",
+          "TASK_STATE_AUTH_REQUIRED",
+      ):
+        return None
+      task_id = metadata.get(A2A_METADATA_PREFIX + "task_id")
+      return task_id if isinstance(task_id, str) else None
+    return None
+
+  def _has_default_response_persistence(self) -> bool:
+    """Whether recorded responses can safely determine task continuation.
+
+    Inbound converters and after-request hooks may suppress a terminal response,
+    leaving an older pause in history. With these overrides, callers retain
+    control of task routing through function responses or request interceptors.
+    """
+    if self._a2a_part_converter is not convert_a2a_part_to_genai_part:
+      return False
+    for name in (
+        "a2a_task_converter",
+        "a2a_status_update_converter",
+        "a2a_message_converter",
+        "a2a_artifact_update_converter",
+        "a2a_part_converter",
+    ):
+      if (
+          getattr(self._config, name)
+          is not A2aRemoteAgentConfig.model_fields[name].default
+      ):
+        return False
+    return not any(
+        interceptor.after_request is not None
+        for interceptor in self._config.request_interceptors or []
+    )
+
   async def _handle_a2a_response(
       self,
       a2a_response: _compat.A2AClientEvent | A2AMessage,
@@ -1447,6 +1528,17 @@ async def _handle_a2a_response(
             for part in event.content.parts or []:
               part.thought = True
           _add_mock_function_call(event, update.status.state)
+        elif (
+            isinstance(update, A2ATaskStatusUpdateEvent)
+            and update.status.state in _TASK_RESPONSE_BOUNDARY_STATES
+        ):
+          # A bare pause or terminal status is still the end of an interaction.
+          # Persist it so the next user turn sees the current task state.
+          event = Event(
+              author=self.name,
+              invocation_id=ctx.invocation_id,
+              branch=ctx.branch,
+          )
         elif isinstance(update, A2ATaskArtifactUpdateEvent):
           # This is a streaming task artifact update.
           # Convert only the parts carried by this update. Converting the
@@ -1667,10 +1759,12 @@ async def _run_async_impl(
             parts=message_parts,
             role=_compat.ROLE_USER,
             context_id=context_id or session_id,
+            task_id=self._get_resumable_task_id(ctx, context_id),
         )
 
       logger.debug(build_a2a_request_log(a2a_request))
 
+      last_task = None
       try:
         intercepted_request, parameters = (
             await execute_before_request_interceptors(
@@ -1697,7 +1791,6 @@ async def _run_async_impl(
         # status/artifact updates are aggregated into a running task (matching the
         # 0.3.x client behavior).
         normalize_stream_item = _compat.make_stream_normalizer()
-        last_task = None
         async with Aclosing(
             _compat.send_message(
                 a2a_client,
@@ -1832,6 +1925,11 @@ async def _run_async_impl(
               error_message=task_error_message,
               invocation_id=ctx.invocation_id,
               branch=ctx.branch,
+              custom_metadata={
+                  **_task_response_metadata(last_task),
+                  A2A_METADATA_PREFIX
+                  + "request": _compat.a2a_to_dict(a2a_request),
+              },
           )
 
       except _compat.A2A_HTTP_ERRORS as e:
@@ -1844,6 +1942,8 @@ async def _run_async_impl(
             A2A_METADATA_PREFIX + "error": error_message,
             A2A_METADATA_PREFIX + "status_code": str(status_code),
         }
+        if last_task is not None:
+          custom_metadata.update(_task_response_metadata(last_task))
         if a2a_request:
           custom_metadata[A2A_METADATA_PREFIX + "request"] = (
               _compat.a2a_to_dict(a2a_request)
@@ -1864,6 +1964,8 @@ async def _run_async_impl(
         custom_metadata = {
             A2A_METADATA_PREFIX + "error": error_message,
         }
+        if last_task is not None:
+          custom_metadata.update(_task_response_metadata(last_task))
         if a2a_request:
           custom_metadata[A2A_METADATA_PREFIX + "request"] = (
               _compat.a2a_to_dict(a2a_request)

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

@@ -55,6 +55,17 @@
     "mock_function_call_for_required_user_auth"
 )
 
+# These states finish the current interaction, including a deliberate pause.
+# Unlike progress updates, a bare boundary still needs a persisted response.
+_TASK_RESPONSE_BOUNDARY_STATES = frozenset({
+    _compat.TS_INPUT_REQUIRED,
+    _compat.TS_AUTH_REQUIRED,
+    _compat.TS_COMPLETED,
+    _compat.TS_FAILED,
+    _compat.TS_CANCELED,
+    _compat.TS_REJECTED,
+})
+
 A2AMessageToEventConverter = Callable[
     [
         Message,
@@ -196,11 +207,21 @@ def _create_event(
     usage_metadata: Any = None,
     error_code: Any = None,
     citation_metadata: Any = None,
+    allow_empty_event: bool = False,
 ) -> Optional[Event]:
   """Creates an ADK event from parts and metadata."""
   event_actions = actions or EventActions()
-  if not output_parts and not event_actions.model_dump(
-      exclude_none=True, exclude_defaults=True
+  if (
+      not allow_empty_event
+      and not output_parts
+      and event_actions == EventActions()
+      and not any((
+          grounding_metadata,
+          custom_metadata,
+          usage_metadata,
+          error_code,
+          citation_metadata,
+      ))
   ):
     return None
 
@@ -496,11 +517,13 @@ def convert_a2a_task_to_event(
     output_parts: list[genai_types.Part] = []
     long_running_function_ids: set[str] = set()
     metadata_fields: dict[str, Any] = {}
+    has_source_parts = False
     status_message = _compat.normalize_message(a2a_task.status.message)
     if a2a_task.artifacts:
       artifact_parts = [
           part for artifact in a2a_task.artifacts for part in artifact.parts
       ]
+      has_source_parts = bool(artifact_parts)
       for artifact in a2a_task.artifacts:
         event_actions = _merge_event_actions(
             event_actions, _extract_event_actions(artifact.metadata)
@@ -523,6 +546,7 @@ def convert_a2a_task_to_event(
       parts, ids = _convert_a2a_parts_to_adk_parts(
           status_message.parts, part_converter
       )
+      has_source_parts = has_source_parts or bool(status_message.parts)
       output_parts.extend(parts)
       long_running_function_ids.update(ids)
     elif status_message and not metadata_fields:
@@ -550,6 +574,10 @@ def convert_a2a_task_to_event(
         author,
         event_actions,
         long_running_function_ids,
+        allow_empty_event=(
+            not has_source_parts
+            and a2a_task.status.state in _TASK_RESPONSE_BOUNDARY_STATES
+        ),
         **metadata_fields,
     )
 
@@ -655,6 +683,10 @@ def convert_a2a_status_update_to_event(
         author,
         event_actions,
         long_running_function_ids,
+        allow_empty_event=(
+            (status_message is None or not status_message.parts)
+            and a2a_status_update.status.state in _TASK_RESPONSE_BOUNDARY_STATES
+        ),
         **metadata_fields,
     )
   except Exception as e: