From af857389d11ef67fc07e5ab93dfb0eee49ce621a Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Sat, 4 Jul 2026 16:21:06 +0530 Subject: [PATCH 1/6] fix(agent_execution): resolve ActiveTask 'destroyed but pending' warning during teardown --- src/a2a/server/agent_execution/active_task.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index b0154c8d6..867a3b26c 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 @@ -574,6 +575,11 @@ async def _run_consumer(self) -> None: self._is_finished.set() self._request_queue.shutdown(immediate=True) await self._event_queue_agent.close(immediate=True) + + if self._producer_task and not self._producer_task.done(): + with contextlib.suppress(asyncio.CancelledError): + await self._producer_task + async with self._lock: self._reference_count -= 1 logger.debug('Consumer[%s]: Finishing', self._task_id) From 99976aeda23c13197e581dcf66de0a5ceb66c654 Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Sun, 5 Jul 2026 19:48:23 +0530 Subject: [PATCH 2/6] fix(agent_execution): explicitly catch asyncio.CancelledError during shutdown --- src/a2a/server/agent_execution/active_task.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index 867a3b26c..583ac2b7a 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -577,8 +577,12 @@ async def _run_consumer(self) -> None: await self._event_queue_agent.close(immediate=True) if self._producer_task and not self._producer_task.done(): - with contextlib.suppress(asyncio.CancelledError): + try: await self._producer_task + except asyncio.CancelledError: + pass + except Exception as e: + logger.debug('Consumer[%s]: Awaited producer_task raised %r', self._task_id, e) async with self._lock: self._reference_count -= 1 From ea135f2b89373ef26c410019deed77ef33ac6b9c Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Sun, 5 Jul 2026 19:54:26 +0530 Subject: [PATCH 3/6] fix(style): resolve ruff linter/formatter line length error --- src/a2a/server/agent_execution/active_task.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index 583ac2b7a..7d37c6d9f 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -582,7 +582,11 @@ async def _run_consumer(self) -> None: except asyncio.CancelledError: pass except Exception as e: - logger.debug('Consumer[%s]: Awaited producer_task raised %r', self._task_id, e) + logger.debug( + 'Consumer[%s]: Awaited producer_task raised %r', + self._task_id, + e, + ) async with self._lock: self._reference_count -= 1 From a9f1a3ad5cc46067dad50f4a5cc3e0521b25a6c9 Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Sun, 5 Jul 2026 19:59:48 +0530 Subject: [PATCH 4/6] fix(style): apply ruff fixes --- src/a2a/server/agent_execution/active_task.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/a2a/server/agent_execution/active_task.py b/src/a2a/server/agent_execution/active_task.py index 7d37c6d9f..718db7500 100644 --- a/src/a2a/server/agent_execution/active_task.py +++ b/src/a2a/server/agent_execution/active_task.py @@ -36,7 +36,6 @@ from __future__ import annotations import asyncio -import contextlib import logging import uuid @@ -581,7 +580,7 @@ async def _run_consumer(self) -> None: await self._producer_task except asyncio.CancelledError: pass - except Exception as e: + except Exception as e: # noqa: BLE001 logger.debug( 'Consumer[%s]: Awaited producer_task raised %r', self._task_id, From ca2fb3273cb833180dcb4ae68363cef0a6645438 Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Sat, 11 Jul 2026 07:52:48 +0530 Subject: [PATCH 5/6] test: add regression test from PR #1125 for issue #1121 --- .../agent_execution/test_active_task.py | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/tests/server/agent_execution/test_active_task.py b/tests/server/agent_execution/test_active_task.py index ce9e2c068..47597c6a0 100644 --- a/tests/server/agent_execution/test_active_task.py +++ b/tests/server/agent_execution/test_active_task.py @@ -895,3 +895,87 @@ 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_awaited_on_normal_completion(): + """Verify producer is awaited before cleanup to prevent GC warnings. + + Regression test for #1121: when the consumer finishes first (normal + completion path), the producer must be 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, + ) + + execute_started = asyncio.Event() + execute_barrier = asyncio.Event() + + async def execute_mock(req, q): + execute_started.set() + 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 + ) + + await execute_started.wait() + + if active_task._consumer_task: + active_task._consumer_task.cancel() + + # Release the producer so it can finish its work and loop back + # to get(), where it will receive QueueShutDown. + execute_barrier.set() + + try: + await active_task._consumer_task + except asyncio.CancelledError: + pass + + await active_task._is_finished.wait() + + 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 83d3de4838d1408624d1527a67fd8e408798ca45 Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Mon, 13 Jul 2026 18:11:05 +0530 Subject: [PATCH 6/6] chore: trigger CI for ITK