feat(server): make V2 event queue capacity configurable

src/a2a/server/agent_execution/active_task.py

@@ -60,6 +60,7 @@
 from a2a.server.cluster.event_stream import VersionedEvent
 from a2a.server.cluster.task_store import ConcurrentTaskModificationError
 from a2a.server.events.event_queue_v2 import (
+    DEFAULT_MAX_QUEUE_SIZE,
     AsyncQueue,
     Event,
     EventQueueSource,
@@ -140,6 +141,9 @@ async def run(self) -> None:
 
                 await self._process_event(event)
 
+        except asyncio.CancelledError as error:
+            self.active_task._consumer_error = error
+            raise
         except QueueShutDown:
             logger.debug(
                 'Consumer[%s]: Event queue shut down',
@@ -152,16 +156,25 @@ async def run(self) -> None:
             if isinstance(e, ConcurrentTaskModificationError):
                 # Other writers kept advancing the task; report a protocol error.
                 error = InternalError(message=str(e))
+            self.active_task._consumer_error = error
 
             updated_task = None
             try:
                 updated_task = await self._write_failed_status()
+            except Exception:
+                logger.exception(
+                    'Consumer[%s]: Failed to persist failure status',
+                    self.active_task._task_id,
+                )
             finally:
                 # Subscribers must always hear about the failure, or the
                 # request waiting on them never returns.
                 await self._enqueue_to_subscribers(
                     cast('Event', error), updated_task
                 )
+        finally:
+            # Validation ends before _run_consumer waits for producer cleanup.
+            self.active_task._response_validation_done.set()
 
     async def _write_failed_status(self) -> Task | None:
         """Records FAILED once a failure ends the event stream.
@@ -296,6 +309,7 @@ def _handle_message_event(self, event: Message) -> None:
                 'Multiple Message objects received.'
             )
         self.task_mode = False
+        self.active_task._message_result = event
 
     async def _handle_task_event(
         self,
@@ -478,6 +492,8 @@ def __init__(  # noqa: PLR0913
         on_cleanup: Callable[[ActiveTask], None] | None = None,
         event_stream: TaskEventStream | None = None,
         interrupted_task_timeout: float | None = None,
+        *,
+        max_event_queue_size: int = DEFAULT_MAX_QUEUE_SIZE,
     ) -> None:
         """Initializes the ActiveTask.
 
@@ -491,15 +507,23 @@ def __init__(  # noqa: PLR0913
                         the task from the ActiveTaskRegistry.
             event_stream: Optional cross-replica stream; applied events are
                         published to it so other replicas can observe them.
+            max_event_queue_size: Positive event capacity for each local event
+                        buffer, including internal lifecycle markers. Applies
+                        to both sources, the agent's default sink and each
+                        subscriber tap, not the pending-request queue or the
+                        cross-replica stream.
             interrupted_task_timeout: Optional seconds after which an idle
                         interrupted task is finished; ``None`` disables it.
         """
         # --- Core Dependencies ---
         self._agent_executor = agent_executor
         self._task_id = task_id
-        self._event_queue_agent = EventQueueSource()
+        self._max_event_queue_size = max_event_queue_size
+        self._event_queue_agent = EventQueueSource(
+            max_queue_size=max_event_queue_size
+        )
         self._event_queue_subscribers = EventQueueSource(
-            create_default_sink=False
+            max_queue_size=max_event_queue_size, create_default_sink=False
         )
         self._task_manager = task_manager
         self._push_sender = push_sender
@@ -519,6 +543,7 @@ def __init__(  # noqa: PLR0913
 
         # _task_created is set when initial version of task is stored in DB.
         self._task_created = asyncio.Event()
+        self._response_validation_done = asyncio.Event()
 
         # `_is_finished` is set EXACTLY ONCE when the consumer loop exits, signifying
         # the absolute end of the task's active lifecycle.
@@ -529,6 +554,8 @@ def __init__(  # noqa: PLR0913
         self._producer_task: asyncio.Task[None] | None = None
         # The background task reading from _event_queue and updating the DB.
         self._consumer_task: asyncio.Task[None] | None = None
+        self._message_result: Message | None = None
+        self._consumer_error: BaseException | None = None
 
         # Tracks how many active SSE/gRPC streams are currently tailing this task.
         # Protected by `_lock`.
@@ -782,7 +809,7 @@ async def subscribe(  # noqa: PLR0915
         # evict_on_full=True: a subscriber that falls behind is dropped, so it
         # cannot wedge dispatch for the others. See _deliver_to_sink().
         tapped_queue = await self._event_queue_subscribers.tap(
-            evict_on_full=True
+            max_queue_size=self._max_event_queue_size, evict_on_full=True
         )
         request_id = await self.enqueue_request(request) if request else None
 
@@ -970,6 +997,13 @@ async def aclose(self) -> None:
         background tasks are cancelled, so teardown is bounded even when a
         subscriber sink was never drained. It is safe to call multiple times.
         """
+        # Immediate queue closure can discard events before validation ends.
+        if (
+            not self._response_validation_done.is_set()
+            and self._consumer_error is None
+        ):
+            self._consumer_error = asyncio.CancelledError()
+
         # Shut down the request queue first, mirroring the producer's
         # ``finally``. If `start()` was never called, the producer/consumer
         # ``finally`` blocks never run, and without this a caller parked in
@@ -1091,6 +1125,23 @@ async def _mark_task_as_failed(self, exception: Exception) -> Task | None:
                 pass
         return task
 
+    async def get_result(self) -> Task | Message:
+        """Returns a task snapshot or a fully validated direct Message result.
+
+        An evicted subscriber may miss the Message or a validation error. Wait
+        for Message validation, independently of producer cleanup, so later
+        invalid events or executor failures still take precedence. Cancelling
+        this waiter does not cancel the shared consumer.
+        """
+        message = self._message_result
+        if message is not None:
+            await self._response_validation_done.wait()
+        if self._consumer_error is not None:
+            raise self._consumer_error
+        if message is not None:
+            return message
+        return await self.get_task()
+
     async def get_task(self) -> Task:
         """Get task from db."""
         await self._task_created.wait()

src/a2a/server/agent_execution/active_task_registry.py

@@ -18,6 +18,7 @@
 
 from a2a.server.agent_execution.active_task import ActiveTask
 from a2a.server.cluster.task_store import StoredTask
+from a2a.server.events.event_queue import DEFAULT_MAX_QUEUE_SIZE
 from a2a.server.tasks.task_manager import TaskManager
 from a2a.utils.errors import TaskNotFoundError
 
@@ -28,19 +29,26 @@
 class ActiveTaskRegistry:
     """A registry for active ActiveTask instances."""
 
-    def __init__(
+    def __init__(  # noqa: PLR0913
         self,
         agent_executor: AgentExecutor,
         task_store: TaskStore | VersionedTaskStore,
         push_sender: PushNotificationSender | None = None,
         event_stream: TaskEventStream | None = None,
         interrupted_task_timeout: float | None = None,
-    ):
+        *,
+        max_event_queue_size: int = DEFAULT_MAX_QUEUE_SIZE,
+    ) -> None:
+        """Initializes a registry with a positive per-buffer event capacity."""
+        if max_event_queue_size <= 0:
+            raise ValueError('max_event_queue_size must be greater than 0')
+
         self._agent_executor = agent_executor
         self._task_store = task_store
         self._push_sender = push_sender
         self._interrupted_task_timeout = interrupted_task_timeout
         self._event_stream = event_stream
+        self._max_event_queue_size = max_event_queue_size
         self._active_tasks: dict[str, ActiveTask] = {}
         self._lock = threading.RLock()
         self._cleanup_tasks: set[asyncio.Task[None]] = set()
@@ -82,6 +90,7 @@ async def get_or_create(
                     on_cleanup=self._on_active_task_cleanup,
                     event_stream=self._event_stream,
                     interrupted_task_timeout=self._interrupted_task_timeout,
+                    max_event_queue_size=self._max_event_queue_size,
                 )
                 self._active_tasks[task_id] = active_task
 

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -24,32 +24,33 @@
     LegacyTaskStoreAdapter,
     VersionedTaskStore,
 )
+from a2a.server.events.event_queue import DEFAULT_MAX_QUEUE_SIZE
 from a2a.server.request_handlers.request_handler import (
     RequestHandler,
     validate,
     validate_request_params,
 )
 from a2a.server.tasks.push_notification_config_store import (
     normalize_push_notification_config,
 )
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
     DeleteTaskPushNotificationConfigRequest,
     GetExtendedAgentCardRequest,
     GetTaskPushNotificationConfigRequest,
     GetTaskRequest,
     ListTaskPushNotificationConfigsRequest,
     ListTaskPushNotificationConfigsResponse,
     ListTasksRequest,
     ListTasksResponse,
     Message,
     SendMessageRequest,
     SubscribeToTaskRequest,
     Task,
     TaskPushNotificationConfig,
     TaskState,
     TaskStatusUpdateEvent,
 )
 from a2a.utils.errors import (
     ExtendedAgentCardNotConfiguredError,
@@ -97,6 +98,22 @@
     to an in-memory `ActiveTaskRegistry`. Passing a non-``None`` value emits a
     `DeprecationWarning` and logs a warning.
 
+    ``max_event_queue_size`` sets the positive event capacity of each local
+    event buffer (default 1024): the two source inputs, the agent's default
+    sink and one sink per local subscription. Internal request-start and
+    request-completion markers also occupy slots. This is a per-buffer event
+    count, not a task memory limit; pending requests, shared storage and remote
+    ``TaskEventStream`` buffers are outside its scope.
+
+    A full default sink backpressures the agent when the internal consumer
+    falls behind. Full subscriber sinks are evicted, including live but slow
+    consumers and the subscriptions used by send and streaming requests.
+    Smaller capacities can end a subscription sooner; a non-streaming send
+    whose subscription is evicted can return a non-terminal stored task
+    snapshot rather than waiting for the agent to finish. Direct Message
+    responses remain available after eviction once the consumer has validated
+    the full response; later validation and executor errors take precedence.
+
     ``interrupted_task_timeout`` releases the `ActiveTask` of a task idle in
     ``input-required`` or ``auth-required`` after that many seconds; a
     follow-up recreates it from the store. ``None`` (default) never releases;
@@ -112,18 +129,20 @@
         agent_card: AgentCard,
         queue_manager: Any
         | None = None,  # Accepted for signature compat; ignored in v2 (warns)
         push_config_store: PushNotificationConfigStore | None = None,
         push_sender: PushNotificationSender | None = None,
         request_context_builder: RequestContextBuilder | None = None,
         extended_agent_card: AgentCard | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], Awaitable[AgentCard]
         ]
         | None = None,
         push_url_validator: Callable[[str], Awaitable[bool]] | None = None,
         validate_input_modes: bool = False,
         event_stream: TaskEventStream | None = None,
         interrupted_task_timeout: float | None = None,
+        *,
+        max_event_queue_size: int = DEFAULT_MAX_QUEUE_SIZE,
     ) -> None:
         if (
             interrupted_task_timeout is not None
@@ -184,6 +203,7 @@
             push_sender=self._push_sender,
             event_stream=self._event_stream,
             interrupted_task_timeout=interrupted_task_timeout,
+            max_event_queue_size=max_event_queue_size,
         )
         self._background_tasks = set()
 
@@ -461,7 +481,7 @@
 
         if result is None:
             logger.debug('Missing result for task %s', request_context.task_id)
-            result = await active_task.get_task()
+            result = await active_task.get_result()
 
         if isinstance(result, Task):
             result = apply_history_length(result, params.configuration)

tests/server/agent_execution/test_active_task.py

@@ -2,6 +2,7 @@
 import contextlib
 import logging
 
+from typing import cast
 from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
@@ -11,6 +12,7 @@
 from a2a.server.agent_execution.agent_executor import AgentExecutor
 from a2a.server.agent_execution.context import RequestContext
 from a2a.server.context import ServerCallContext
+from a2a.server.events.event_queue import Event
 from a2a.server.events.event_queue_v2 import EventQueueSource as EventQueue
 from a2a.server.tasks.push_notification_sender import PushNotificationSender
 from a2a.server.tasks.task_manager import TaskManager
@@ -1336,3 +1338,69 @@ async def slow_execute(req, q):
     for t in background:
         t.cancel()
     await asyncio.gather(*background, return_exceptions=True)
+
+
+@pytest.mark.timeout(5)
+@pytest.mark.asyncio
+@pytest.mark.parametrize('max_event_queue_size', [1, 2])
+async def test_small_event_queue_capacity_evicts_slow_subscriber(
+    max_event_queue_size: int,
+) -> None:
+    task_manager = Mock(spec=TaskManager)
+    task_manager.get_task = AsyncMock(
+        return_value=Task(
+            id='test-task-id',
+            status=TaskStatus(state=TaskState.TASK_STATE_WORKING),
+        )
+    )
+    active_task = ActiveTask(
+        agent_executor=Mock(spec=AgentExecutor),
+        task_id='test-task-id',
+        task_manager=task_manager,
+        max_event_queue_size=max_event_queue_size,
+    )
+    active_task._task_created.set()
+    source = active_task._event_queue_subscribers
+    slow = active_task.subscribe(include_initial_task=True)
+    fast = active_task.subscribe(include_initial_task=True)
+    subscriptions = [slow, fast]
+    readers: list[asyncio.Task[Event]] = []
+    try:
+        assert isinstance(await anext(slow), Task)
+        slow_sink = next(iter(source._sinks))
+        assert isinstance(await anext(fast), Task)
+
+        for index in range(max_event_queue_size + 2):
+            reader = asyncio.create_task(anext(fast))
+            readers.append(reader)
+            event = Task(id=f'event-{index}')
+            await source.enqueue_event(cast('Event', (event, None)))
+            await source.test_only_join_incoming_queue()
+            assert await reader == event
+            assert slow_sink.is_closed() == (index >= max_event_queue_size)
+
+        assert slow_sink not in source._sinks
+        assert len(source._sinks) == 1
+
+        late = active_task.subscribe(include_initial_task=True)
+        subscriptions.append(late)
+        assert isinstance(await anext(late), Task)
+        fast_reader = asyncio.create_task(anext(fast))
+        late_reader = asyncio.create_task(anext(late))
+        readers.extend((fast_reader, late_reader))
+        event = Task(id='after-eviction')
+        await source.enqueue_event(cast('Event', (event, None)))
+        await source.test_only_join_incoming_queue()
+        assert await fast_reader == event
+        assert await late_reader == event
+    finally:
+        for reader in readers:
+            if not reader.done():
+                reader.cancel()
+        await asyncio.gather(*readers, return_exceptions=True)
+        for subscription in subscriptions:
+            await subscription.aclose()
+        await active_task.aclose()
+
+    assert not source._sinks
+    assert source._dispatcher_task.done()

tests/server/agent_execution/test_active_task_registry.py

@@ -37,6 +37,18 @@ def _make_registry() -> ActiveTaskRegistry:
     )
 
 
+@pytest.mark.parametrize('max_event_queue_size', [0, -1])
+def test_registry_rejects_nonpositive_event_queue_size(
+    max_event_queue_size: int,
+) -> None:
+    with pytest.raises(ValueError, match='max_event_queue_size'):
+        ActiveTaskRegistry(
+            agent_executor=_SlowExecutor(),
+            task_store=InMemoryTaskStore(),
+            max_event_queue_size=max_event_queue_size,
+        )
+
+
 @pytest.mark.timeout(5)
 @pytest.mark.asyncio
 async def test_aclose_reaps_active_tasks_and_empties_registry():

tests/server/cluster/conftest.py

@@ -18,7 +18,7 @@
     VersionedTaskStore,
 )
 from a2a.server.context import ServerCallContext
-from a2a.server.events.event_queue import Event
+from a2a.server.events.event_queue import DEFAULT_MAX_QUEUE_SIZE, Event
 from a2a.server.owner_resolver import OwnerResolver, resolve_user_scope
 from a2a.server.request_handlers.default_request_handler_v2 import (
     DefaultRequestHandlerV2,
@@ -390,13 +390,15 @@ def make_replica(
     store: VersionedTaskStore,
     stream: TaskEventStream,
     agent: AgentExecutor,
+    max_event_queue_size: int = DEFAULT_MAX_QUEUE_SIZE,
 ) -> DefaultRequestHandlerV2:
     """Builds a handler ('replica') wired to a shared store and stream."""
     return DefaultRequestHandlerV2(
         agent_executor=agent,
         task_store=store,
         agent_card=streaming_agent_card(),
         event_stream=stream,
+        max_event_queue_size=max_event_queue_size,
     )