fix: mark failed A2A tasks as error events

Author: sakiko-toyokawa

src/google/adk/a2a/_compat.py

@@ -139,6 +139,9 @@ def _proto_to_dict(msg: Any) -> dict[str, Any]:
   TS_AUTH_REQUIRED = TaskState.Value("TASK_STATE_AUTH_REQUIRED")
   TS_CANCELED = TaskState.Value("TASK_STATE_CANCELED")
 
+  A2A_TASK_FAILED_ERROR_CODE = "A2A_TASK_FAILED"
+  A2A_TASK_FAILED_ERROR_MESSAGE = "Remote agent task failed"
+
   TP_JSONRPC = TransportProtocol.JSONRPC
   TP_HTTP_JSON = TransportProtocol.HTTP_JSON
   TP_GRPC = TransportProtocol.GRPC
@@ -154,6 +157,9 @@ def _proto_to_dict(msg: Any) -> dict[str, Any]:
   TS_AUTH_REQUIRED = TaskState.auth_required
   TS_CANCELED = TaskState.canceled
 
+  A2A_TASK_FAILED_ERROR_CODE = "A2A_TASK_FAILED"
+  A2A_TASK_FAILED_ERROR_MESSAGE = "Remote agent task failed"
+
   TP_JSONRPC = getattr(TransportProtocol, "jsonrpc")
   TP_HTTP_JSON = getattr(TransportProtocol, "http_json")
   TP_GRPC = getattr(TransportProtocol, "grpc")
@@ -1187,6 +1193,11 @@ def role_to_str(role: Any) -> str:
   return "user" if role == ROLE_USER else "model"
 
 
+def is_failed_status(status: Any) -> bool:
+  """Returns whether an A2A task status is in the failed state."""
+  return status is not None and getattr(status, "state", None) == TS_FAILED
+
+
 def normalize_message(msg: Any) -> Any:
   """Collapses an empty 1.x proto ``Message`` to ``None``.
 

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

@@ -52,6 +52,23 @@
 logger = logging.getLogger("google_adk." + __name__)
 
 
+def _extract_text_from_event(event: Event) -> str:
+  """Returns the text content of an ADK event, or an empty string."""
+  if event.content and event.content.parts:
+    return "".join(part.text or "" for part in event.content.parts if part.text)
+  return ""
+
+
+def _mark_a2a_task_failed(event: Event) -> Event:
+  """Marks an event produced from a failed A2A task."""
+  event.error_code = _compat.A2A_TASK_FAILED_ERROR_CODE
+  if not event.error_message:
+    event.error_message = (
+        _extract_text_from_event(event) or _compat.A2A_TASK_FAILED_ERROR_MESSAGE
+    )
+  return event
+
+
 AdkEventToA2AEventsConverter = Callable[
     [
         Event,
@@ -250,27 +267,30 @@ def convert_a2a_task_to_event(
       if agent_messages:
         message = agent_messages[-1]
 
-    # Convert message if available
+    # Convert message if available; otherwise create a minimal event.
     if message:
       try:
         event: Event = convert_a2a_message_to_event(
             message, author, invocation_context, part_converter=part_converter
         )
-        return event
       except Exception as e:
         logger.error("Failed to convert A2A task message to event: %s", e)
         raise RuntimeError(f"Failed to convert task message: {e}") from e
+    else:
+      event = Event(
+          invocation_id=(
+              invocation_context.invocation_id
+              if invocation_context
+              else platform_uuid.new_uuid()
+          ),
+          author=author or "a2a agent",
+          branch=invocation_context.branch if invocation_context else None,
+      )
 
-    # Create minimal event if no message is available
-    return Event(
-        invocation_id=(
-            invocation_context.invocation_id
-            if invocation_context
-            else platform_uuid.new_uuid()
-        ),
-        author=author or "a2a agent",
-        branch=invocation_context.branch if invocation_context else None,
-    )
+    if _compat.is_failed_status(a2a_task.status):
+      event = _mark_a2a_task_failed(event)
+
+    return event
 
   except Exception as e:
     logger.error("Failed to convert A2A task to event: %s", e)

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

@@ -195,12 +195,15 @@ def _create_event(
     custom_metadata: Any = None,
     usage_metadata: Any = None,
     error_code: Any = None,
+    error_message: Any = None,
     citation_metadata: Any = None,
 ) -> 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 output_parts
+      and not error_code
+      and not event_actions.model_dump(exclude_none=True, exclude_defaults=True)
   ):
     return None
 
@@ -229,12 +232,30 @@ def _create_event(
       custom_metadata=custom_metadata,
       usage_metadata=usage_metadata,
       error_code=error_code,
+      error_message=error_message,
       citation_metadata=citation_metadata,
   )
 
   return event
 
 
+def _extract_text_from_event(event: Event) -> str:
+  """Returns the text content of an ADK event, or an empty string."""
+  if event.content and event.content.parts:
+    return "".join(part.text or "" for part in event.content.parts if part.text)
+  return ""
+
+
+def _mark_a2a_task_failed(event: Event) -> Event:
+  """Marks an event produced from a failed A2A task."""
+  event.error_code = _compat.A2A_TASK_FAILED_ERROR_CODE
+  if not event.error_message:
+    event.error_message = (
+        _extract_text_from_event(event) or _compat.A2A_TASK_FAILED_ERROR_MESSAGE
+    )
+  return event
+
+
 def _a2a_role_to_content_role(role: Optional[Role]) -> str:
   """Maps an A2A Role to the corresponding GenAI content role."""
   return _compat.role_to_str(role)
@@ -513,6 +534,7 @@ def convert_a2a_task_to_event(
     if status_message and (
         a2a_task.status.state == _compat.TS_INPUT_REQUIRED
         or a2a_task.status.state == _compat.TS_AUTH_REQUIRED
+        or _compat.is_failed_status(a2a_task.status)
     ):
       event_actions = _merge_event_actions(
           event_actions,
@@ -534,14 +556,24 @@ def convert_a2a_task_to_event(
         )
     )
 
-    return _create_event(
+    event = _create_event(
         output_parts,
         invocation_context,
         author,
         event_actions,
         long_running_function_ids,
         **metadata_fields,
     )
+    if _compat.is_failed_status(a2a_task.status):
+      if event is None:
+        event = _create_event(
+            [],
+            invocation_context,
+            author,
+            error_code=_compat.A2A_TASK_FAILED_ERROR_CODE,
+        )
+      event = _mark_a2a_task_failed(event)
+    return event
 
   except Exception as e:
     logger.error("Failed to convert A2A task to event: %s", e)
@@ -581,14 +613,15 @@ def convert_a2a_message_to_event(
     )
     content_role = _a2a_role_to_content_role(getattr(a2a_message, "role", None))
     metadata_fields = _extract_all_metadata_fields(a2a_message.metadata)
-    return _create_event(
+    event = _create_event(
         output_parts,
         invocation_context,
         author,
         _extract_event_actions(a2a_message.metadata),
         content_role=content_role,
         **metadata_fields,
     )
+    return event
 
   except Exception as e:
     logger.error("Failed to convert A2A message to event: %s", e)
@@ -639,14 +672,24 @@ def convert_a2a_status_update_to_event(
         )
     )
 
-    return _create_event(
+    event = _create_event(
         output_parts,
         invocation_context,
         author,
         event_actions,
         long_running_function_ids,
         **metadata_fields,
     )
+    if _compat.is_failed_status(a2a_status_update.status):
+      if event is None:
+        event = _create_event(
+            [],
+            invocation_context,
+            author,
+            error_code=_compat.A2A_TASK_FAILED_ERROR_CODE,
+        )
+      event = _mark_a2a_task_failed(event)
+    return event
   except Exception as e:
     logger.error("Failed to convert A2A status update to event: %s", e)
     raise RuntimeError(f"Failed to convert status update: {e}") from e

src/google/adk/agents/remote_a2a_agent.py

@@ -122,6 +122,30 @@
 })
 
 
+def _mark_a2a_task_failed_event(
+    event: Optional[Event],
+    *,
+    author: str,
+    invocation_context: InvocationContext,
+) -> Event:
+  """Marks an event produced from a failed A2A task, creating one if needed."""
+  if event is None:
+    event = Event(
+        author=author,
+        invocation_id=invocation_context.invocation_id,
+        branch=invocation_context.branch,
+    )
+  event.error_code = _compat.A2A_TASK_FAILED_ERROR_CODE
+  if not event.error_message:
+    text = "".join(
+        part.text or ""
+        for part in (event.content.parts if event.content else [])
+        if part.text
+    )
+    event.error_message = text or _compat.A2A_TASK_FAILED_ERROR_MESSAGE
+  return event
+
+
 def _payload_is_auth_config(payload: Any) -> bool:
   """Whether a payload looks like a serialized AuthConfig (fail closed)."""
   candidate = payload
@@ -774,7 +798,16 @@ async def _handle_a2a_response(
               task, self.name, ctx, self._a2a_part_converter
           )
           if not event:
-            return None
+            if _compat.is_failed_status(getattr(task, "status", None)):
+              event = _mark_a2a_task_failed_event(
+                  None, author=self.name, invocation_context=ctx
+              )
+            else:
+              return None
+          elif _compat.is_failed_status(getattr(task, "status", None)):
+            event = _mark_a2a_task_failed_event(
+                event, author=self.name, invocation_context=ctx
+            )
           # for streaming task, we update the event with the task status.
           # We update the event as Thought updates.
           if (
@@ -791,29 +824,54 @@ async def _handle_a2a_response(
             for part in event.content.parts or []:
               part.thought = True
           _add_mock_function_call(event, task.status.state)
-        elif isinstance(update, A2ATaskStatusUpdateEvent) and (
-            _status_message := (
-                _compat.normalize_message(update.status.message)
-                if update.status
-                else None
-            )
-        ):
-          # This is a streaming task status update with a message.
+        elif isinstance(update, A2ATaskStatusUpdateEvent):
           # ``normalize_message`` collapses the always-present empty proto
-          # ``Message`` (1.x) to ``None`` so this branch only fires when a real
-          # message is attached, matching 0.3.x where the field is ``None``.
-          event = convert_a2a_message_to_event(
-              _status_message, self.name, ctx, self._a2a_part_converter
+          # ``Message`` (1.x) to ``None`` so status updates without a real
+          # message are handled explicitly below, matching 0.3.x.
+          _status_message = (
+              _compat.normalize_message(update.status.message)
+              if update.status
+              else None
           )
-          if not event:
+          failed_state = (
+              _compat.TS_FAILED
+              if _compat.is_failed_status(getattr(update, "status", None))
+              or _compat.is_failed_status(getattr(task, "status", None))
+              else None
+          )
+          if failed_state is not None:
+            event = (
+                convert_a2a_message_to_event(
+                    _status_message,
+                    self.name,
+                    ctx,
+                    self._a2a_part_converter,
+                )
+                if _status_message
+                else None
+            )
+            event = _mark_a2a_task_failed_event(
+                event, author=self.name, invocation_context=ctx
+            )
+            _add_mock_function_call(event, failed_state)
+          elif _status_message:
+            # This is a streaming task status update with a message.
+            event = convert_a2a_message_to_event(
+                _status_message, self.name, ctx, self._a2a_part_converter
+            )
+            if not event:
+              return None
+            if event.content is not None and update.status.state in (
+                _compat.TS_SUBMITTED,
+                _compat.TS_WORKING,
+            ):
+              for part in event.content.parts or []:
+                part.thought = True
+            _add_mock_function_call(event, update.status.state)
+          else:
+            # This is a streaming status update without a message (e.g. status
+            # change). We don't emit an event for non-failed updates.
             return None
-          if event.content is not None and update.status.state in (
-              _compat.TS_SUBMITTED,
-              _compat.TS_WORKING,
-          ):
-            for part in event.content.parts or []:
-              part.thought = True
-          _add_mock_function_call(event, update.status.state)
         elif isinstance(update, A2ATaskArtifactUpdateEvent):
           # This is a streaming task artifact update.
           # Convert only the parts carried by this update. Converting the
@@ -904,11 +962,28 @@ async def _handle_a2a_response_v2(
           event = self._config.a2a_task_converter(
               task, self.name, ctx, self._config.a2a_part_converter
           )
+          if not event:
+            if _compat.is_failed_status(getattr(task, "status", None)):
+              event = _mark_a2a_task_failed_event(
+                  None, author=self.name, invocation_context=ctx
+              )
+            else:
+              return None
+          elif _compat.is_failed_status(getattr(task, "status", None)):
+            event = _mark_a2a_task_failed_event(
+                event, author=self.name, invocation_context=ctx
+            )
         elif isinstance(update, A2ATaskStatusUpdateEvent):
           # This is a streaming task status update.
           event = self._config.a2a_status_update_converter(
               update, self.name, ctx, self._config.a2a_part_converter
           )
+          if _compat.is_failed_status(getattr(update, "status", None)) or (
+              _compat.is_failed_status(getattr(task, "status", None))
+          ):
+            event = _mark_a2a_task_failed_event(
+                event, author=self.name, invocation_context=ctx
+            )
         elif isinstance(update, A2ATaskArtifactUpdateEvent):
           # This is a streaming task artifact update.
           event = self._config.a2a_artifact_update_converter(