From 5b9821db970d6823b23a1d72a110be1391e89279 Mon Sep 17 00:00:00 2001
From: isheng <ishengeqi@163.com>
Date: Wed, 8 Jul 2026 02:50:01 +0800
Subject: [PATCH 1/4] 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 <ishengeqi@163.com>
Date: Wed, 8 Jul 2026 02:54:19 +0800
Subject: [PATCH 2/4] =?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 <ishengeqi@163.com>
Date: Wed, 8 Jul 2026 03:07:25 +0800
Subject: [PATCH 3/4] 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,
+        )

From cc16ff44c99d42be2866ab6f5ae69dbae3d98ee6 Mon Sep 17 00:00:00 2001
From: isheng <ishengeqi@163.com>
Date: Wed, 8 Jul 2026 03:11:14 +0800
Subject: [PATCH 4/4] fix(server): save and restore call_context in cancel() to
 prevent race with producer

The cancel() method was directly setting self._task_manager._call_context,
which races with the producer loop that also sets _call_context for
incoming requests. When both run concurrently, the shared _call_context
can be overwritten, causing task_store operations to use the wrong
authorization context.

Fix: save the original _call_context before cancel() operations and
restore it in a finally block. This resolves the TODO at line ~512
documenting the conflict.
---
 src/a2a/server/agent_execution/active_task.py | 72 ++++++++++---------
 .../agent_execution/test_active_task.py       | 62 ++++++++++++++++
 2 files changed, 101 insertions(+), 33 deletions(-)

diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py
index 0633a4415..6b2b5094b 100644
--- a/src/a2a/server/agent_execution/active_task.py
+++ b/src/a2a/server/agent_execution/active_task.py
@@ -701,43 +701,49 @@ async def cancel(self, call_context: ServerCallContext) -> Task:
         """
         logger.debug('Cancel[%s]: Cancelling task', self._task_id)
 
-        # TODO: Conflicts with call_context on the pending request.
-        self._task_manager._call_context = call_context
-
-        task = await self._task_manager.get_task()
-        request_context = RequestContext(
-            call_context=call_context,
-            task_id=self._task_id,
-            context_id=task.context_id if task else None,
-            task=task,
-        )
+        # Save and restore _call_context to avoid races with the producer
+        # loop which also sets _call_context for incoming requests.
+        # See the TODO at the producer loop line ~512.
+        saved_call_context = self._task_manager._call_context
+        try:
+            self._task_manager._call_context = call_context
+
+            task = await self._task_manager.get_task()
+            request_context = RequestContext(
+                call_context=call_context,
+                task_id=self._task_id,
+                context_id=task.context_id if task else None,
+                task=task,
+            )
 
-        async with self._lock:
-            if not self._is_finished.is_set() and self._producer_task:
-                logger.debug(
-                    'Cancel[%s]: Cancelling producer task', self._task_id
-                )
-                self._producer_task.cancel()
-                try:
-                    await self._agent_executor.cancel(
-                        request_context, self._event_queue_agent
+            async with self._lock:
+                if not self._is_finished.is_set() and self._producer_task:
+                    logger.debug(
+                        'Cancel[%s]: Cancelling producer task', self._task_id
                     )
-                except Exception as e:
-                    logger.exception(
-                        'Cancel[%s]: Agent cancel failed', self._task_id
+                    self._producer_task.cancel()
+                    try:
+                        await self._agent_executor.cancel(
+                            request_context, self._event_queue_agent
+                        )
+                    except Exception as e:
+                        logger.exception(
+                            'Cancel[%s]: Agent cancel failed', self._task_id
+                        )
+                        await self._mark_task_as_failed(e)
+                        raise
+                else:
+                    logger.debug(
+                        'Cancel[%s]: Task already finished [%s] or producer not started [%s], not cancelling',
+                        self._task_id,
+                        self._is_finished.is_set(),
+                        self._producer_task,
                     )
-                    await self._mark_task_as_failed(e)
-                    raise
-            else:
-                logger.debug(
-                    'Cancel[%s]: Task already finished [%s] or producer not started [%s], not cancelling',
-                    self._task_id,
-                    self._is_finished.is_set(),
-                    self._producer_task,
-                )
 
-        await self._is_finished.wait()
-        task = await self._task_manager.get_task()
+            await self._is_finished.wait()
+            task = await self._task_manager.get_task()
+        finally:
+            self._task_manager._call_context = saved_call_context
         if not task:
             raise RuntimeError('Task should have been created')
         return task
diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py
index ca6f89ff9..04568e0e8 100644
--- a/tests/server/agent_execution/test_active_task.py
+++ b/tests/server/agent_execution/test_active_task.py
@@ -987,3 +987,65 @@ async def execute_mock(req, q):
     assert cleanup_called, (
         'on_cleanup should be called after producer is drained'
     )
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+async def test_cancel_restores_call_context():
+    """Verify cancel() restores _call_context after completion.
+
+    Regression test: cancel() temporarily sets _call_context to its
+    own call_context for task operations, then restores the original
+    value to prevent races with the producer loop.
+    """
+    agent_executor = Mock()
+    task_manager = Mock()
+    task_manager.get_task = AsyncMock()
+    original_context = ServerCallContext()
+    task_manager._call_context = original_context
+
+    active_task = ActiveTask(
+        agent_executor=agent_executor,
+        task_id='test-task-id',
+        task_manager=task_manager,
+    )
+
+    stop_event = asyncio.Event()
+
+    async def execute_mock(req, q):
+        await stop_event.wait()
+
+    agent_executor.execute = AsyncMock(side_effect=execute_mock)
+    agent_executor.cancel = AsyncMock()
+    task_manager.get_task.side_effect = [
+        Task(
+            id='test-task-id',
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+    ] * 20
+    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 = original_context
+    request_context.context_id = 'test-context'
+    request_context.message = None
+
+    await active_task.enqueue_request(request_context)
+    await active_task.start(
+        call_context=original_context, create_task_if_missing=True
+    )
+    await asyncio.sleep(0.1)
+
+    cancel_context = ServerCallContext()
+    await active_task.cancel(cancel_context)
+    stop_event.set()
+
+    # Verify call_context was restored
+    assert task_manager._call_context is original_context