fix(server): Improve event consumer error handling (#282)

# Description
Enhances the event consumer to gracefully handle
`pydantic.ValidationError` by logging the error and continuing execution
instead of crashing. This improves the reliability of the event stream
processing.

**Changes:**
- `src/a2a/server/events/event_consumer.py`: Added specific `except`
block for `ValidationError`.
- `tests/server/events/test_event_consumer.py`: Added a test case to
verify the new error handling logic.

Author: pankaj-bind

src/a2a/server/events/event_consumer.py

@@ -4,6 +4,8 @@
 
 from collections.abc import AsyncGenerator
 
+from pydantic import ValidationError
+
 from a2a.server.events.event_queue import Event, EventQueue
 from a2a.types import (
     InternalError,
@@ -138,6 +140,9 @@ async def consume_all(self) -> AsyncGenerator[Event]:
                 # python 3.12 and get a queue empty error on an open queue
                 if self.queue.is_closed():
                     break
+            except ValidationError as e:
+                logger.error(f"Invalid event format received: {e}")
+                continue
             except Exception as e:
                 logger.error(
                     f'Stopping event consumption due to exception: {e}'

tests/server/events/test_event_consumer.py

@@ -1,9 +1,9 @@
 import asyncio
-
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
+from pydantic import ValidationError
 
 from a2a.server.events.event_consumer import EventConsumer, QueueClosed
 from a2a.server.events.event_queue import EventQueue
@@ -343,3 +343,24 @@ def test_agent_task_callback_no_exception(event_consumer: EventConsumer):
 
     assert event_consumer._exception is None  # Should remain None
     mock_task.exception.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_consume_all_handles_validation_error(
+    event_consumer: EventConsumer, mock_event_queue: AsyncMock
+):
+    """Test that consume_all gracefully handles a pydantic.ValidationError."""
+    # Simulate dequeue_event raising a ValidationError
+    mock_event_queue.dequeue_event.side_effect = [
+        ValidationError.from_exception_data(title="Test Error", line_errors=[]),
+        asyncio.CancelledError  # To stop the loop for the test
+    ]
+
+    with patch("a2a.server.events.event_consumer.logger.error") as logger_error_mock:
+        with pytest.raises(asyncio.CancelledError):
+            async for _ in event_consumer.consume_all():
+                pass
+
+        # Check that the specific error was logged and the consumer continued
+        logger_error_mock.assert_called_once()
+        assert "Invalid event format received" in logger_error_mock.call_args[0][0]