From 06b9b22a07af6a32d9c11c8bb5170ae1fde766c1 Mon Sep 17 00:00:00 2001 From: sakiko-toyokawa Date: Thu, 13 Aug 2026 18:39:42 +0800 Subject: [PATCH] fix: mark failed A2A tasks as error events --- src/google/adk/a2a/_compat.py | 11 ++ .../adk/a2a/converters/event_converter.py | 44 +++++-- src/google/adk/a2a/converters/to_adk_event.py | 53 +++++++- src/google/adk/agents/remote_a2a_agent.py | 117 ++++++++++++++---- .../a2a/converters/test_event_converter.py | 38 ++++++ tests/unittests/a2a/converters/test_to_adk.py | 52 ++++++++ .../unittests/agents/test_remote_a2a_agent.py | 108 +++++++++++++++- 7 files changed, 384 insertions(+), 39 deletions(-) diff --git a/src/google/adk/a2a/_compat.py b/src/google/adk/a2a/_compat.py index d5c0c2d820c..00fa884ec2a 100644 --- a/src/google/adk/a2a/_compat.py +++ b/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``. diff --git a/src/google/adk/a2a/converters/event_converter.py b/src/google/adk/a2a/converters/event_converter.py index 2e6904b16bd..22016affbf1 100644 --- a/src/google/adk/a2a/converters/event_converter.py +++ b/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) diff --git a/src/google/adk/a2a/converters/to_adk_event.py b/src/google/adk/a2a/converters/to_adk_event.py index e03f8a597c3..2f37cfe59fa 100644 --- a/src/google/adk/a2a/converters/to_adk_event.py +++ b/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,7 +556,7 @@ def convert_a2a_task_to_event( ) ) - return _create_event( + event = _create_event( output_parts, invocation_context, author, @@ -542,6 +564,16 @@ def convert_a2a_task_to_event( 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,7 +613,7 @@ 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, @@ -589,6 +621,7 @@ def convert_a2a_message_to_event( content_role=content_role, **metadata_fields, ) + return event except Exception as e: logger.error("Failed to convert A2A message to event: %s", e) @@ -639,7 +672,7 @@ def convert_a2a_status_update_to_event( ) ) - return _create_event( + event = _create_event( output_parts, invocation_context, author, @@ -647,6 +680,16 @@ def convert_a2a_status_update_to_event( 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 diff --git a/src/google/adk/agents/remote_a2a_agent.py b/src/google/adk/agents/remote_a2a_agent.py index bc171335f40..d79766cbdb1 100644 --- a/src/google/adk/agents/remote_a2a_agent.py +++ b/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( diff --git a/tests/unittests/a2a/converters/test_event_converter.py b/tests/unittests/a2a/converters/test_event_converter.py index 3e7a3658d6f..4c1c79e71cb 100644 --- a/tests/unittests/a2a/converters/test_event_converter.py +++ b/tests/unittests/a2a/converters/test_event_converter.py @@ -827,6 +827,44 @@ def test_convert_a2a_task_to_event_no_message(self): assert result.branch == "test-branch" assert result.invocation_id == "test-invocation-id" + def test_convert_a2a_task_to_event_failed_status_sets_error(self): + """Test failed A2A task status maps to an error Event.""" + failed_message = _compat.make_message( + message_id="msg-failed", + role=_compat.ROLE_AGENT, + parts=[_compat.make_text_part("Remote agent task failed")], + ) + task = _compat.make_task( + id="task-failed", + status=_compat.make_task_status( + _compat.TS_FAILED, message=failed_message + ), + ) + + result = convert_a2a_task_to_event( + task, "test-author", self.mock_invocation_context + ) + + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + assert result.content is not None + assert result.content.parts[0].text == "Remote agent task failed" + + def test_convert_a2a_task_to_event_failed_status_without_message(self): + """Test failed A2A task without a message still emits an error event.""" + task = _compat.make_task( + id="task-failed", + status=_compat.make_task_status(_compat.TS_FAILED), + ) + + result = convert_a2a_task_to_event( + task, "test-author", self.mock_invocation_context + ) + + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + assert result.content is None + @patch("google.adk.a2a.converters.event_converter.platform_uuid.new_uuid") def test_convert_a2a_task_to_event_default_author(self, mock_uuid): """Test converting A2A task with default author and no invocation context.""" diff --git a/tests/unittests/a2a/converters/test_to_adk.py b/tests/unittests/a2a/converters/test_to_adk.py index 75e5329a42d..035307c40fa 100644 --- a/tests/unittests/a2a/converters/test_to_adk.py +++ b/tests/unittests/a2a/converters/test_to_adk.py @@ -178,6 +178,38 @@ def test_convert_a2a_task_to_event_success(self): assert len(event.content.parts) == 1 assert event.content.parts[0] == mock_genai_part + def test_convert_a2a_task_to_event_failed_status_sets_error(self): + """Test failed A2A task status maps to an error Event.""" + a2a_part = _make_a2a_part_for_test({}) + failed_message = Message( + message_id="msg-failed", + role=_compat.ROLE_AGENT, + parts=[a2a_part], + ) + task = _compat.make_task( + id="task-failed", + status=_compat.make_task_status( + _compat.TS_FAILED, message=failed_message + ), + ) + mock_part_converter = Mock( + return_value=[ + genai_types.Part.from_text(text="Remote agent task failed") + ] + ) + + event = convert_a2a_task_to_event( + task, + author="test-author", + invocation_context=self.mock_context, + part_converter=mock_part_converter, + ) + + assert event is not None + assert event.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert event.error_message == "Remote agent task failed" + assert event.content.parts[0].text == "Remote agent task failed" + def test_convert_a2a_task_to_event_returns_action_only_event(self): """Test A2A task conversion returns action-only events.""" task = Task( @@ -634,6 +666,26 @@ def test_convert_a2a_status_update_to_event_success(self): assert len(event.content.parts) == 1 assert event.content.parts[0] == mock_genai_part + def test_convert_a2a_status_update_to_event_failed_status_without_message( + self, + ): + """Test failed status update without a message still emits an error event.""" + update = _compat.make_task_status_update_event( + task_id="task-failed", + status=_compat.make_task_status(_compat.TS_FAILED), + context_id="context-1", + final=False, + ) + + event = convert_a2a_status_update_to_event( + update, "test-author", self.mock_context, Mock() + ) + + assert event is not None + assert event.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert event.error_message == "Remote agent task failed" + assert event.content is None + def test_convert_a2a_status_update_to_event_none(self): """Test convert_a2a_status_update_to_event with None.""" with pytest.raises(ValueError, match="A2A status update cannot be None"): diff --git a/tests/unittests/agents/test_remote_a2a_agent.py b/tests/unittests/agents/test_remote_a2a_agent.py index e423e33d889..033450c05ab 100644 --- a/tests/unittests/agents/test_remote_a2a_agent.py +++ b/tests/unittests/agents/test_remote_a2a_agent.py @@ -28,7 +28,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 @@ -1510,6 +1509,40 @@ async def test_handle_a2a_response_with_task_completed_and_no_update(self): assert A2A_METADATA_PREFIX + "task_id" in result.custom_metadata assert A2A_METADATA_PREFIX + "context_id" in result.custom_metadata + @pytest.mark.asyncio + async def test_handle_a2a_response_failed_task_sets_error(self): + """Test non-streaming failed A2A task maps to an error event.""" + mock_a2a_task = Mock(spec=A2ATask) + mock_a2a_task.id = "task-123" + mock_a2a_task.context_id = "context-123" + mock_a2a_task.status = Mock(spec=A2ATaskStatus) + mock_a2a_task.status.state = _compat.TS_FAILED + + mock_event = Event( + author=self.agent.name, + invocation_id=self.mock_context.invocation_id, + branch=self.mock_context.branch, + content=genai_types.Content( + role="model", + parts=[genai_types.Part.from_text(text="Remote agent task failed")], + ), + ) + + with patch.object( + remote_a2a_agent, + "convert_a2a_task_to_event", + autospec=True, + ) as mock_convert: + mock_convert.return_value = mock_event + + result = await self.agent._handle_a2a_response( + (mock_a2a_task, None), self.mock_context + ) + + assert result is not None + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + def test_construct_message_parts_from_session_preserves_order(self): """Test that message parts are in correct order with multi-part messages. @@ -1828,6 +1861,27 @@ async def test_handle_a2a_response_with_task_status_update_no_message(self): assert result is None + @pytest.mark.asyncio + async def test_handle_a2a_response_failed_status_update_without_message(self): + """Test failed streaming status without a message still emits an error.""" + mock_a2a_task = Mock(spec=A2ATask) + mock_a2a_task.id = "task-123" + mock_a2a_task.context_id = "context-123" + + mock_update = Mock(spec=TaskStatusUpdateEvent) + mock_update.status = Mock(spec=A2ATaskStatus) + mock_update.status.state = _compat.TS_FAILED + mock_update.status.message = None + + result = await self.agent._handle_a2a_response( + (mock_a2a_task, mock_update), self.mock_context + ) + + assert result is not None + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + assert result.content is None + @pytest.mark.asyncio async def test_handle_a2a_response_with_artifact_update(self): """Test successful A2A response handling with artifact update.""" @@ -2502,6 +2556,34 @@ async def test_handle_a2a_response_impl_with_task_and_no_update(self): == "context-123" ) + @pytest.mark.asyncio + async def test_handle_a2a_response_impl_failed_task_sets_error(self): + """Test v2 non-streaming failed task maps to an error event.""" + mock_a2a_task = Mock(spec=A2ATask) + mock_a2a_task.id = "task-123" + mock_a2a_task.context_id = None + mock_a2a_task.status = Mock(spec=A2ATaskStatus) + mock_a2a_task.status.state = _compat.TS_FAILED + + mock_event = Event( + author=self.agent.name, + invocation_id=self.mock_context.invocation_id, + branch=self.mock_context.branch, + content=genai_types.Content( + role="model", + parts=[genai_types.Part.from_text(text="Remote agent task failed")], + ), + ) + self.mock_config.a2a_task_converter.return_value = mock_event + + result = await self.agent._handle_a2a_response_v2( + (mock_a2a_task, None), self.mock_context + ) + + assert result is not None + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + @pytest.mark.asyncio async def test_handle_a2a_response_impl_with_task_status_update(self): """Test _handle_a2a_response_impl with TaskStatusUpdateEvent.""" @@ -2534,6 +2616,30 @@ async def test_handle_a2a_response_impl_with_task_status_update(self): assert result.custom_metadata[A2A_METADATA_PREFIX + "task_id"] == "task-123" assert A2A_METADATA_PREFIX + "context_id" not in result.custom_metadata + @pytest.mark.asyncio + async def test_handle_a2a_response_impl_failed_status_update_returns_error( + self, + ): + """Test v2 failed status update returns an error even if converter is None.""" + mock_a2a_task = Mock(spec=A2ATask) + mock_a2a_task.id = "task-123" + mock_a2a_task.context_id = None + + mock_update = Mock(spec=TaskStatusUpdateEvent) + mock_update.status = Mock(spec=A2ATaskStatus) + mock_update.status.state = _compat.TS_FAILED + + self.mock_config.a2a_status_update_converter.return_value = None + + result = await self.agent._handle_a2a_response_v2( + (mock_a2a_task, mock_update), self.mock_context + ) + + assert result is not None + assert result.error_code == _compat.A2A_TASK_FAILED_ERROR_CODE + assert result.error_message == "Remote agent task failed" + assert result.content is None + @pytest.mark.asyncio async def test_handle_a2a_response_impl_with_task_artifact_update(self): """Test _handle_a2a_response_impl with TaskArtifactUpdateEvent."""