From 17fce50eb54b48fc20b0d39637466f6902c33a7c Mon Sep 17 00:00:00 2001 From: andy Date: Fri, 9 Oct 2026 20:13:12 +0800 Subject: [PATCH 1/3] fix(server): isolate push notification failures from the task lifecycle (#1313) A failure in the push-notification infrastructure (config store read or URL validator) propagated into the EventConsumer's generic failure path and rewrote an otherwise successful task as TASK_STATE_FAILED, failing message/send instead of returning the task result. Two non-breaking guards: - BasePushNotificationSender.send_notification: read get_info_for_dispatch inside an error boundary and treat a raising push_url_validator as a rejected URL, both degrading to 'notification not sent' with an exception log. - EventConsumer._update_task_state: wrap the send_notification call in try/except so a sender failure is logged and never disturbs the task state or the event stream. Fixes #1313 Signed-off-by: andy --- src/a2a/server/agent_execution/active_task.py | 15 ++++-- .../tasks/base_push_notification_sender.py | 33 +++++++++--- .../test_default_request_handler_v2.py | 54 +++++++++++++++++++ .../tasks/test_push_notification_sender.py | 40 ++++++++++++++ 4 files changed, 133 insertions(+), 9 deletions(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index a2f174367..6c845ab6d 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -346,9 +346,18 @@ async def _update_task_state( 'Consumer[%s]: Sending push notification', self.active_task._task_id, ) - await self.active_task._push_sender.send_notification( - self.active_task._task_id, event - ) + try: + await self.active_task._push_sender.send_notification( + self.active_task._task_id, event + ) + except Exception: + # Push delivery is best-effort: a sender failure must not + # fail the task or disturb the event stream. + logger.exception( + 'Consumer[%s]: Push notification sender raised; ' + 'ignoring the failure.', + self.active_task._task_id, + ) async def _handle_terminal_state(self, updated_task: Task) -> None: logger.debug( diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 62c9ccaa2..d9c51e95f 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -69,7 +69,20 @@ async def send_notification( self, task_id: str, event: PushNotificationEvent ) -> None: """Sends a push notification for an event if configuration exists.""" - push_configs = await self._config_store.get_info_for_dispatch(task_id) + try: + push_configs = await self._config_store.get_info_for_dispatch( + task_id + ) + except Exception: + # Push delivery is best-effort: an infrastructure failure while + # reading the config store must not propagate into the task + # lifecycle. + logger.exception( + 'Failed to read push notification configs for task_id=%s; ' + 'skipping push notification delivery.', + task_id, + ) + return if not push_configs: return @@ -91,11 +104,19 @@ async def _dispatch_notification( task_id: str, ) -> bool: url = push_info.url - if ( - self._push_url_validator is not None - and not await self._push_url_validator(url) - ): - return False + if self._push_url_validator is not None: + try: + accepted = await self._push_url_validator(url) + except Exception: + logger.exception( + 'Push URL validator raised for task_id=%s, URL: %s. ' + 'Treating the URL as rejected.', + task_id, + url, + ) + return False + if not accepted: + return False try: headers: dict[str, str] = {} if push_info.token: diff --git a/tests/server/request_handlers/test_default_request_handler_v2.py b/tests/server/request_handlers/test_default_request_handler_v2.py index b2d25f95a..f789051a5 100644 --- a/tests/server/request_handlers/test_default_request_handler_v2.py +++ b/tests/server/request_handlers/test_default_request_handler_v2.py @@ -7,6 +7,7 @@ from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from a2a.auth.user import UnauthenticatedUser, User @@ -33,6 +34,9 @@ TaskStore, TaskUpdater, ) +from a2a.server.tasks.base_push_notification_sender import ( + BasePushNotificationSender, +) from a2a.server.tasks.task_manager import TaskManager from a2a.types import ( ContentTypeNotSupportedError, @@ -2591,3 +2595,53 @@ def test_init_does_not_warn_for_complete_agent_cards( extended_agent_card=_complete_agent_card(), ) assert caplog.records == [] + + +class FlakyPushConfigStore(InMemoryPushNotificationConfigStore): + """Config store whose dispatch read fails transiently (issue #1313).""" + + async def get_info_for_dispatch( + self, task_id: str + ) -> list[TaskPushNotificationConfig]: + raise RuntimeError('transient DB error while reading push configs') + + +@pytest.mark.asyncio +async def test_push_config_store_failure_does_not_fail_task(): + """A push-notification infrastructure failure must not rewrite a + successfully completed task as FAILED, and message/send must still + return the task result (#1313).""" + task_store = InMemoryTaskStore() + request_handler = DefaultRequestHandlerV2( + agent_executor=HelloAgentExecutor(), + task_store=task_store, + push_config_store=FlakyPushConfigStore(), + push_sender=BasePushNotificationSender( + httpx_client=httpx.AsyncClient(), + config_store=FlakyPushConfigStore(), + ), + agent_card=create_default_agent_card(), + ) + params = SendMessageRequest( + message=Message( + role=Role.ROLE_USER, + message_id='msg_push_store_fail', + parts=[Part(text='Hi')], + ), + configuration=SendMessageConfiguration( + accepted_output_modes=['text/plain'] + ), + ) + + result = await request_handler.on_message_send( + params, create_server_call_context() + ) + + assert isinstance(result, Task) + assert result.status.state == TaskState.TASK_STATE_COMPLETED + get_task_result = await request_handler.on_get_task( + GetTaskRequest(id=result.id), create_server_call_context() + ) + assert get_task_result is not None + assert isinstance(get_task_result, Task) + assert get_task_result.status.state == TaskState.TASK_STATE_COMPLETED diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index 0c87a6d4e..afc6f982e 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -313,6 +313,46 @@ async def test_send_notification_artifact_update_event(self) -> None: headers={}, ) + @patch('a2a.server.tasks.base_push_notification_sender.logger') + async def test_send_notification_config_store_failure_is_swallowed( + self, mock_logger: MagicMock + ) -> None: + """A config store read failure is logged and delivery is skipped + instead of propagating into the task lifecycle (issue #1313).""" + task_id = 'task_store_error' + task_data = _create_sample_task(task_id=task_id) + self.mock_config_store.get_info_for_dispatch.side_effect = RuntimeError( + 'transient DB error' + ) + + await self.sender.send_notification(task_id, task_data) + + self.mock_config_store.get_info_for_dispatch.assert_awaited_once_with( + task_id + ) + self.mock_httpx_client.post.assert_not_called() + mock_logger.exception.assert_called_once() + + @patch('a2a.server.tasks.base_push_notification_sender.logger') + async def test_push_url_validator_raising_rejects_url( + self, mock_logger: MagicMock + ) -> None: + """A raising push_url_validator is treated as a rejected URL rather + than propagating into the task lifecycle (issue #1313).""" + sender = BasePushNotificationSender( + httpx_client=self.mock_httpx_client, + config_store=self.mock_config_store, + push_url_validator=AsyncMock(side_effect=RuntimeError('boom')), + ) + task_data = _create_sample_task(task_id='task_validator_error') + config = _create_sample_push_config(url='http://notify.me/here') + self.mock_config_store.get_info_for_dispatch.return_value = [config] + + await sender.send_notification(task_data.id, task_data) + + self.mock_httpx_client.post.assert_not_called() + mock_logger.exception.assert_called_once() + def _gai_result(ip: str, port: int = 80): return [(2, 1, 6, '', (ip, port))] From 7c6741885016ad2de9b8ac7212909ea25dab2d1e Mon Sep 17 00:00:00 2001 From: andy Date: Fri, 9 Oct 2026 20:19:37 +0800 Subject: [PATCH 2/3] fix(server): guard the legacy handler push path and harden push-failure tests Extends the #1313 fix to LegacyRequestHandler: its _send_push_notification_if_needed had the same unguarded sender call in both the blocking and streaming paths, so a push-infrastructure failure could fail the task there as well. Also, per self-review: - cover the consumer-level guard in EventConsumer._update_task_state with a raising custom PushNotificationSender (defense-in-depth branch was previously untested); - mirror the codebase's mocked-httpx-client pattern in the v2 e2e tests; - document the fail-closed semantics of the push URL validator and the defense-in-depth role of the consumer guard. Signed-off-by: andy --- src/a2a/server/agent_execution/active_task.py | 5 +- .../default_request_handler.py | 11 ++++- .../tasks/base_push_notification_sender.py | 2 +- .../test_default_request_handler.py | 24 ++++++++++ .../test_default_request_handler_v2.py | 47 ++++++++++++++++++- 5 files changed, 85 insertions(+), 4 deletions(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index 6c845ab6d..ab35a6bda 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -352,7 +352,10 @@ async def _update_task_state( ) except Exception: # Push delivery is best-effort: a sender failure must not - # fail the task or disturb the event stream. + # fail the task or disturb the event stream. This guard is + # defense-in-depth on top of the boundary inside + # BasePushNotificationSender, and also protects against + # custom PushNotificationSender implementations. logger.exception( 'Consumer[%s]: Push notification sender raised; ' 'ignoring the failure.', diff --git a/src/a2a/server/request_handlers/default_request_handler.py b/src/a2a/server/request_handlers/default_request_handler.py index 209d693a7..7fe0e43c0 100644 --- a/src/a2a/server/request_handlers/default_request_handler.py +++ b/src/a2a/server/request_handlers/default_request_handler.py @@ -382,7 +382,16 @@ async def _send_push_notification_if_needed( and task_id and isinstance(event, PushNotificationEvent) ): - await self._push_sender.send_notification(task_id, event) + try: + await self._push_sender.send_notification(task_id, event) + except Exception: + # Push delivery is best-effort: a sender failure must not + # fail the task or disturb the event stream. + logger.exception( + 'Push notification sender raised for task_id=%s; ' + 'ignoring the failure.', + task_id, + ) @validate_request_params async def on_message_send( diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index d9c51e95f..86cca309f 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -110,7 +110,7 @@ async def _dispatch_notification( except Exception: logger.exception( 'Push URL validator raised for task_id=%s, URL: %s. ' - 'Treating the URL as rejected.', + 'Failing closed and treating the URL as rejected.', task_id, url, ) diff --git a/tests/server/request_handlers/test_default_request_handler.py b/tests/server/request_handlers/test_default_request_handler.py index 925019900..4338d2586 100644 --- a/tests/server/request_handlers/test_default_request_handler.py +++ b/tests/server/request_handlers/test_default_request_handler.py @@ -3451,3 +3451,27 @@ def test_init_does_not_warn_for_complete_agent_cards( extended_agent_card=_complete_agent_card(), ) assert caplog.records == [] + + +@pytest.mark.asyncio +async def test_push_sender_failure_does_not_propagate(agent_card): + """A push-notification infrastructure failure raised by the sender must + not propagate out of _send_push_notification_if_needed into the task + lifecycle (#1313).""" + mock_push_sender = AsyncMock(spec=PushNotificationSender) + mock_push_sender.send_notification.side_effect = RuntimeError( + 'transient push infrastructure error' + ) + request_handler = DefaultRequestHandler( + agent_executor=AsyncMock(spec=AgentExecutor), + task_store=AsyncMock(spec=TaskStore), + push_config_store=AsyncMock(spec=PushNotificationConfigStore), + push_sender=mock_push_sender, + agent_card=agent_card, + ) + event = create_sample_task() + + # Must not raise: push delivery is best-effort. + await request_handler._send_push_notification_if_needed(event.id, event) + + mock_push_sender.send_notification.assert_awaited_once_with(event.id, event) diff --git a/tests/server/request_handlers/test_default_request_handler_v2.py b/tests/server/request_handlers/test_default_request_handler_v2.py index f789051a5..50aaca220 100644 --- a/tests/server/request_handlers/test_default_request_handler_v2.py +++ b/tests/server/request_handlers/test_default_request_handler_v2.py @@ -2606,6 +2606,13 @@ async def get_info_for_dispatch( raise RuntimeError('transient DB error while reading push configs') +class RaisingPushSender(PushNotificationSender): + """A custom (public-interface) sender that always fails (issue #1313).""" + + async def send_notification(self, task_id, event) -> None: + raise RuntimeError('sender exploded') + + @pytest.mark.asyncio async def test_push_config_store_failure_does_not_fail_task(): """A push-notification infrastructure failure must not rewrite a @@ -2617,7 +2624,7 @@ async def test_push_config_store_failure_does_not_fail_task(): task_store=task_store, push_config_store=FlakyPushConfigStore(), push_sender=BasePushNotificationSender( - httpx_client=httpx.AsyncClient(), + httpx_client=AsyncMock(spec=httpx.AsyncClient), config_store=FlakyPushConfigStore(), ), agent_card=create_default_agent_card(), @@ -2645,3 +2652,41 @@ async def test_push_config_store_failure_does_not_fail_task(): assert get_task_result is not None assert isinstance(get_task_result, Task) assert get_task_result.status.state == TaskState.TASK_STATE_COMPLETED + + +@pytest.mark.asyncio +async def test_custom_push_sender_failure_does_not_fail_task(): + """A raising custom PushNotificationSender must not rewrite a + completed task as FAILED — the consumer-level guard is + defense-in-depth behind the BasePushNotificationSender boundary + (#1313).""" + task_store = InMemoryTaskStore() + request_handler = DefaultRequestHandlerV2( + agent_executor=HelloAgentExecutor(), + task_store=task_store, + push_sender=RaisingPushSender(), + agent_card=create_default_agent_card(), + ) + params = SendMessageRequest( + message=Message( + role=Role.ROLE_USER, + message_id='msg_push_sender_fail', + parts=[Part(text='Hi')], + ), + configuration=SendMessageConfiguration( + accepted_output_modes=['text/plain'] + ), + ) + + result = await request_handler.on_message_send( + params, create_server_call_context() + ) + + assert isinstance(result, Task) + assert result.status.state == TaskState.TASK_STATE_COMPLETED + get_task_result = await request_handler.on_get_task( + GetTaskRequest(id=result.id), create_server_call_context() + ) + assert get_task_result is not None + assert isinstance(get_task_result, Task) + assert get_task_result.status.state == TaskState.TASK_STATE_COMPLETED From f80494c00eb9cf2f3aa5632e01463838c96b1e60 Mon Sep 17 00:00:00 2001 From: andy Date: Fri, 9 Oct 2026 21:47:06 +0800 Subject: [PATCH 3/3] docs(server): document the best-effort contract of push notification senders State the guarantee introduced by the #1313 guards on the interface and base implementation: implementations may raise, the framework catches and logs, and failures never affect the task lifecycle or the event stream. Signed-off-by: andy --- src/a2a/server/tasks/base_push_notification_sender.py | 6 +++++- src/a2a/server/tasks/push_notification_sender.py | 7 ++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 86cca309f..070361a52 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -68,7 +68,11 @@ def __init__( async def send_notification( self, task_id: str, event: PushNotificationEvent ) -> None: - """Sends a push notification for an event if configuration exists.""" + """Sends a push notification for an event if configuration exists. + + Best-effort by design: failures to read the configuration store or + to deliver a notification are logged and do not raise. + """ try: push_configs = await self._config_store.get_info_for_dispatch( task_id diff --git a/src/a2a/server/tasks/push_notification_sender.py b/src/a2a/server/tasks/push_notification_sender.py index 95fa43b69..679f59102 100644 --- a/src/a2a/server/tasks/push_notification_sender.py +++ b/src/a2a/server/tasks/push_notification_sender.py @@ -17,4 +17,9 @@ class PushNotificationSender(ABC): async def send_notification( self, task_id: str, event: PushNotificationEvent ) -> None: - """Sends a push notification containing the latest task state.""" + """Sends a push notification containing the latest task state. + + Implementations are treated as best-effort: the framework catches + and logs any exception raised here, so a failure never affects the + task lifecycle or the event stream. + """