feat: Support AgentExectuor enqueue of a Task object. (#960)

Fixes Task object handling when using new DefaultRequestHandlerV2.

Fixes #869🦕

Author: bartek-w

src/a2a/server/agent_execution/active_task.py

@@ -391,11 +391,22 @@ async def _run_consumer(self) -> None:  # noqa: PLR0915, PLR0912
                                     )
 
                                 if isinstance(event, Task):
-                                    new_task = event
-                                    await self._task_manager.save_task_event(
-                                        new_task
+                                    existing_task = (
+                                        await self._task_manager.get_task()
                                     )
-                                    # TODO: Avoid duplicated messages
+                                    if existing_task:
+                                        logger.error(
+                                            'Task %s already exists. Ignoring task replacement.',
+                                            self._task_id,
+                                        )
+                                    else:
+                                        await (
+                                            self._task_manager.save_task_event(
+                                                event
+                                            )
+                                        )
+                                    # Initial task should already contain the message.
+                                    message_to_save = None
                                 else:
                                     new_task = (
                                         await self._task_manager.ensure_task_id(

tests/integration/test_end_to_end.py

@@ -13,7 +13,7 @@
 from a2a.server.agent_execution import AgentExecutor, RequestContext
 from a2a.server.events import EventQueue
 from a2a.server.events.in_memory_queue_manager import InMemoryQueueManager
-from a2a.server.request_handlers import GrpcHandler, LegacyRequestHandler
+from a2a.server.request_handlers import GrpcHandler, DefaultRequestHandler
 from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
 from a2a.server.routes.rest_routes import create_rest_routes
 from a2a.server.tasks import TaskUpdater
@@ -174,8 +174,7 @@ class ClientSetup(NamedTuple):
 @pytest.fixture
 def base_e2e_setup(agent_card):
     task_store = InMemoryTaskStore()
-    # TODO(https://github.com/a2aproject/a2a-python/issues/869): Use DefaultRequestHandler once it's fixed
-    handler = LegacyRequestHandler(
+    handler = DefaultRequestHandler(
         agent_executor=MockAgentExecutor(),
         task_store=task_store,
         agent_card=agent_card,

tests/integration/test_scenarios.py

@@ -1558,3 +1558,87 @@ async def cancel(
             'Bug: Task should include the published artifact'
         )
         assert last_event.task.artifacts[0].artifact_id == 'art-1'
+
+
+# Scenario: Enqueue Task twice
+@pytest.mark.timeout(2.0)
+@pytest.mark.asyncio
+@pytest.mark.parametrize('use_legacy', [False, True], ids=['v2', 'legacy'])
+@pytest.mark.parametrize(
+    'streaming', [False, True], ids=['blocking', 'streaming']
+)
+async def test_scenario_enqueue_task_twice(caplog, use_legacy, streaming):
+    class DoubleTaskAgent(AgentExecutor):
+        async def execute(
+            self, context: RequestContext, event_queue: EventQueue
+        ):
+            task1 = Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(
+                    state=TaskState.TASK_STATE_WORKING,
+                    message=Message(parts=[Part(text='First task')]),
+                ),
+            )
+            await event_queue.enqueue_event(task1)
+
+            # This is undefined behavior, but it should not crash or hang.
+            task2 = Task(
+                id=context.task_id,
+                context_id=context.context_id,
+                status=TaskStatus(
+                    state=TaskState.TASK_STATE_WORKING,
+                    message=Message(parts=[Part(text='Second task')]),
+                ),
+            )
+            await event_queue.enqueue_event(task2)
+
+            await event_queue.enqueue_event(
+                TaskStatusUpdateEvent(
+                    task_id=context.task_id,
+                    context_id=context.context_id,
+                    status=TaskStatus(state=TaskState.TASK_STATE_COMPLETED),
+                )
+            )
+
+        async def cancel(
+            self, context: RequestContext, event_queue: EventQueue
+        ):
+            pass
+
+    handler = create_handler(DoubleTaskAgent(), use_legacy)
+    client = await create_client(
+        handler, agent_card=agent_card(), streaming=streaming
+    )
+
+    msg = Message(
+        message_id='test-msg', role=Role.ROLE_USER, parts=[Part(text='start')]
+    )
+
+    it = client.send_message(
+        SendMessageRequest(
+            message=msg,
+            configuration=SendMessageConfiguration(return_immediately=False),
+        )
+    )
+    events = [event async for event in it]
+
+    (final_task,) = (await client.list_tasks(ListTasksRequest())).tasks
+
+    if use_legacy:
+        assert [part.text for part in final_task.history[0].parts] == [
+            'Second task'
+        ]
+    else:
+        assert [part.text for part in final_task.history[0].parts] == [
+            'First task'
+        ]
+
+        # Validate that new version logs with error exactly once 'Ignoring task replacement'
+        error_logs = [
+            record.message
+            for record in caplog.records
+            if record.levelname == 'ERROR'
+            and 'Ignoring task replacement' in record.message
+        ]
+        assert len(error_logs) == 1