From 5b9821db970d6823b23a1d72a110be1391e89279 Mon Sep 17 00:00:00 2001 From: isheng Date: Wed, 8 Jul 2026 02:50:01 +0800 Subject: [PATCH 1/3] fix(server): cancel and await producer task before cleanup to prevent GC warning (Fixes #1121) On the normal-completion teardown path, _run_consumer's finally block was releasing the ActiveTask reference (via _maybe_cleanup) before the producer task had finished, leaving it parked on _request_queue.get(). When GC reclaimed the ActiveTask, the still-pending producer was destroyed, triggering asyncio 'Task was destroyed but it is pending!' warnings. Fix: cancel _producer_task and await it (suppressing CancelledError) before calling _maybe_cleanup(), mirroring what cancel() already does on the explicit cancel path. This ensures deterministic teardown and prevents the GC warning. --- src/a2a/server/agent_execution/active_task.py | 13 +++ .../agent_execution/test_active_task.py | 92 +++++++++++++++++++ 2 files changed, 105 insertions(+) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index b0154c8d6..3ae547345 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -36,6 +36,7 @@ from __future__ import annotations import asyncio +import contextlib import logging import uuid @@ -577,6 +578,18 @@ async def _run_consumer(self) -> None: async with self._lock: self._reference_count -= 1 logger.debug('Consumer[%s]: Finishing', self._task_id) + + # Cancel and await the producer before cleanup to prevent + # asyncio "Task was destroyed but it is pending!" warnings. + # Without this, the producer could still be parked on + # _request_queue.get() or inside its finally block when + # _maybe_cleanup() releases the ActiveTask reference, + # causing the still-pending task to be GC'd. + if self._producer_task and not self._producer_task.done(): + self._producer_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._producer_task + await self._maybe_cleanup() async def subscribe( diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index ce9e2c068..ca6f89ff9 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -895,3 +895,95 @@ async def execute_mock(req, q): assert len(events) == 0 await active_task.cancel(request_context) + + +@pytest.mark.timeout(5) +@pytest.mark.asyncio +async def test_producer_cancelled_on_normal_completion(): + """Verify producer is cancelled and awaited before cleanup to prevent GC warnings. + + Regression test for #1121: when the consumer finishes first (normal + completion path), the producer must be cancelled and awaited before + _maybe_cleanup() releases the ActiveTask reference, otherwise asyncio + logs "Task was destroyed but it is pending!". + """ + agent_executor = Mock() + task_manager = Mock() + cleanup_called = False + + def on_cleanup(_task: ActiveTask) -> None: + nonlocal cleanup_called + cleanup_called = True + + active_task = ActiveTask( + agent_executor=agent_executor, + task_id='test-task-id', + task_manager=task_manager, + on_cleanup=on_cleanup, + ) + + # Simulate a producer that blocks waiting for a request. + # The consumer will finish first (after processing all events), + # triggering normal teardown. + execute_started = asyncio.Event() + execute_barrier = asyncio.Event() + + async def execute_mock(req, q): + execute_started.set() + # Block until released — simulates producer waiting for next request + await execute_barrier.wait() + + agent_executor.execute = AsyncMock(side_effect=execute_mock) + agent_executor.cancel = AsyncMock() + task_manager.get_task = AsyncMock( + return_value=Task( + id='test-task-id', + status=TaskStatus(state=TaskState.TASK_STATE_WORKING), + ) + ) + task_manager.save_task_event = AsyncMock() + task_manager.ensure_task_id = AsyncMock( + return_value=Task( + id='test-task-id', + status=TaskStatus(state=TaskState.TASK_STATE_WORKING), + ) + ) + task_manager.process = AsyncMock(side_effect=lambda x: x) + + request_context = Mock(spec=RequestContext) + request_context.call_context = ServerCallContext() + request_context.context_id = 'test-context' + request_context.message = None + + await active_task.enqueue_request(request_context) + await active_task.start( + call_context=ServerCallContext(), create_task_if_missing=True + ) + + # Wait for producer to be executing + await execute_started.wait() + + # Cancel the consumer to force consumer teardown while producer is pending. + # This simulates the normal-completion race: consumer finishes first, + # triggering _maybe_cleanup while producer is still parked. + if active_task._consumer_task: + active_task._consumer_task.cancel() + try: + await active_task._consumer_task + except asyncio.CancelledError: + pass + + # Release the producer so it can complete its CancelledError/finally path + execute_barrier.set() + + # Wait for producer to finish + await active_task._is_finished.wait() + + # Verify producer was properly completed, not left pending + assert active_task._producer_task is not None + assert active_task._producer_task.done(), ( + 'Producer task should be done after consumer teardown' + ) + assert cleanup_called, ( + 'on_cleanup should be called after producer is drained' + ) From afed3afb7708f1e684d2491e856cab19097dac17 Mon Sep 17 00:00:00 2001 From: isheng Date: Wed, 8 Jul 2026 02:54:19 +0800 Subject: [PATCH 2/3] =?UTF-8?q?chore:=20fix=20spelling=20check=20failure?= =?UTF-8?q?=20=E2=80=94=20avoid=20GC'd=20in=20comment,=20add=20GC=20to=20a?= =?UTF-8?q?llow=20list?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/actions/spelling/allow.txt | 1 + src/a2a/server/agent_execution/active_task.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/actions/spelling/allow.txt b/.github/actions/spelling/allow.txt index 1701f14ef..399075923 100644 --- a/.github/actions/spelling/allow.txt +++ b/.github/actions/spelling/allow.txt @@ -150,4 +150,5 @@ typ typeerror UIDs vulnz +GC whl diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index 3ae547345..0633a4415 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -584,7 +584,7 @@ async def _run_consumer(self) -> None: # Without this, the producer could still be parked on # _request_queue.get() or inside its finally block when # _maybe_cleanup() releases the ActiveTask reference, - # causing the still-pending task to be GC'd. + # causing the still-pending task to be garbage-collected. if self._producer_task and not self._producer_task.done(): self._producer_task.cancel() with contextlib.suppress(asyncio.CancelledError): From fbb262b46e890d070021c40bb6a382e80f40be41 Mon Sep 17 00:00:00 2001 From: isheng Date: Wed, 8 Jul 2026 03:07:25 +0800 Subject: [PATCH 3/3] feat(server): send Authorization header in push notifications when auth is configured (Fixes #585) The A2A protocol spec requires servers to authenticate push notifications when the client provides an authentication scheme in PushNotificationConfig. Previously BasePushNotificationSender only sent the X-A2A-Notification-Token header and completely ignored the authentication field. Now the _dispatch_notification method checks push_info.HasField('authentication') and adds an Authorization header with '{scheme} {credentials}' when both scheme and credentials are non-empty. This supports Bearer and any other scheme the client provides. --- .../tasks/base_push_notification_sender.py | 15 ++- .../tasks/test_push_notification_sender.py | 111 ++++++++++++++++++ 2 files changed, 125 insertions(+), 1 deletion(-) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index ff9ca3ce5..1257c6601 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -82,10 +82,23 @@ async def _dispatch_notification( ) -> bool: url = push_info.url try: - headers = None + headers: dict[str, str] | None = None if push_info.token: headers = {'X-A2A-Notification-Token': push_info.token} + # Add Authorization header when authentication is configured. + # The A2A spec requires servers to authenticate push notifications + # when the client provides an authentication scheme in + # PushNotificationConfig. + if push_info.HasField('authentication'): + auth = push_info.authentication + scheme = auth.scheme + credentials = auth.credentials + if scheme and credentials: + if headers is None: + headers = {} + headers['Authorization'] = f'{scheme} {credentials}' + response = await self._client.post( url, json=MessageToDict(to_stream_response(event)), diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index 990f6c7f5..7a0945e0e 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -8,6 +8,7 @@ BasePushNotificationSender, ) from a2a.types.a2a_pb2 import ( + AuthenticationInfo, StreamResponse, Task, TaskArtifactUpdateEvent, @@ -228,3 +229,113 @@ async def test_send_notification_artifact_update_event(self) -> None: json=MessageToDict(StreamResponse(artifact_update=event)), headers=None, ) + + async def test_send_notification_with_bearer_auth(self) -> None: + """Push notification includes Authorization header for Bearer scheme.""" + task_id = 'task_bearer_auth' + task_data = _create_sample_task(task_id=task_id) + config = TaskPushNotificationConfig( + id='cfg1', + url='http://notify.me/here', + authentication=AuthenticationInfo( + scheme='Bearer', + credentials='secret-token-abc', + ), + ) + self.mock_config_store.get_info_for_dispatch.return_value = [config] + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.status_code = 200 + self.mock_httpx_client.post.return_value = mock_response + + await self.sender.send_notification(task_id, task_data) + + self.mock_httpx_client.post.assert_awaited_once_with( + config.url, + json=MessageToDict(StreamResponse(task=task_data)), + headers={'Authorization': 'Bearer secret-token-abc'}, + ) + + async def test_send_notification_with_token_and_bearer_auth(self) -> None: + """Both X-A2A-Notification-Token and Authorization headers are sent.""" + task_id = 'task_token_and_auth' + task_data = _create_sample_task(task_id=task_id) + config = TaskPushNotificationConfig( + id='cfg1', + url='http://notify.me/here', + token='notification-token-xyz', + authentication=AuthenticationInfo( + scheme='Bearer', + credentials='secret-token-abc', + ), + ) + self.mock_config_store.get_info_for_dispatch.return_value = [config] + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.status_code = 200 + self.mock_httpx_client.post.return_value = mock_response + + await self.sender.send_notification(task_id, task_data) + + self.mock_httpx_client.post.assert_awaited_once_with( + config.url, + json=MessageToDict(StreamResponse(task=task_data)), + headers={ + 'X-A2A-Notification-Token': 'notification-token-xyz', + 'Authorization': 'Bearer secret-token-abc', + }, + ) + + async def test_send_notification_auth_no_scheme_no_header(self) -> None: + """Authentication with empty scheme should not add Authorization header.""" + task_id = 'task_auth_no_scheme' + task_data = _create_sample_task(task_id=task_id) + config = TaskPushNotificationConfig( + id='cfg1', + url='http://notify.me/here', + authentication=AuthenticationInfo( + scheme='', + credentials='secret', + ), + ) + self.mock_config_store.get_info_for_dispatch.return_value = [config] + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.status_code = 200 + self.mock_httpx_client.post.return_value = mock_response + + await self.sender.send_notification(task_id, task_data) + + self.mock_httpx_client.post.assert_awaited_once_with( + config.url, + json=MessageToDict(StreamResponse(task=task_data)), + headers=None, + ) + + async def test_send_notification_auth_no_credentials_no_header( + self, + ) -> None: + """Authentication with empty credentials should not add Authorization header.""" + task_id = 'task_auth_no_creds' + task_data = _create_sample_task(task_id=task_id) + config = TaskPushNotificationConfig( + id='cfg1', + url='http://notify.me/here', + authentication=AuthenticationInfo( + scheme='Bearer', + credentials='', + ), + ) + self.mock_config_store.get_info_for_dispatch.return_value = [config] + + mock_response = AsyncMock(spec=httpx.Response) + mock_response.status_code = 200 + self.mock_httpx_client.post.return_value = mock_response + + await self.sender.send_notification(task_id, task_data) + + self.mock_httpx_client.post.assert_awaited_once_with( + config.url, + json=MessageToDict(StreamResponse(task=task_data)), + headers=None, + )