[Bug]: EventQueueSource dispatch wedges when one sink stops draining

[astrogilda]

Splitting this out of #1101 per @rohityan's request there ("the EventQueueSource dispatch layer needs its own targeted architectural improvement. So we would prefer that issue to tracked separately"). #1101 / #1105 cover shutdown teardown. This issue is the runtime wedge @dzacball reproduced on a live server.

Symptom

Under a burst of concurrent non-blocking message/send requests that complete around the same time, the server stops serving entirely, including the static /.well-known/agent-card.json route. CPU idle, nothing logged, and it never recovers without a restart. Full details and a two-file repro (repro_server.py + repro_hammer.py, needs only a2a-sdk[http-server], uvicorn, and httpx) are in #1101, starting at dzacball's comment.

Mechanism

The chain is confined to EventQueueSource._dispatch_loop in src/a2a/server/events/event_queue_v2.py:

1. A subscriber sink stops being drained because its consumer was abandoned. For non-blocking sends the HTTP response returns while the sink lives on.
2. The sink's bounded queue (default 1024) fills.
3. _dispatch_loop runs asyncio.gather(sink._put_internal(event) for every sink), so the full sink's await self._queue.put(event) blocks the gather forever — and dispatch to every other sink stops with it.
4. The source's _incoming_queue fills behind the stalled dispatcher. Every producer now wedges in enqueue_event, and close(immediate=False) can never finish joining.
5. Once every in-flight task reaches this state, the loop has nothing runnable and parks in select(). New connections are never accepted.

A task dump during the wedge shows exactly this shape (from #1101): ~40 EventQueueSource._dispatch_loop, ~20 each of _run_producer / _run_consumer, and one EventQueueSink._put_internal parked on a full sink.

The docstring on EventQueueSource.close() already names this class of problem ("deadlock risk if there are unconsumed events in any of the child sinks and the consumer has crashed without draining its queue"). At runtime, though, the blocking happens in dispatch, before close is ever involved, and one bad sink takes down delivery to all the healthy ones.

Proposed fix: evict-on-full for subscriber taps

The two kinds of sink want different fullness behavior. The default sink is flow control — its consumer drives task state, so blocking dispatch on a full queue is correct backpressure and should stay. Tapped subscriber sinks are broadcast observers whose remote consumers can be abandoned (a non-blocking send whose response already returned), so for them a full queue means the consumer is gone, not slow.

Concretely: tap() gains evict_on_full: bool = False. When an evict-on-full sink is full at delivery time, the dispatcher force-closes it (close(immediate=True), warning logged) and it detaches through the existing remove_sink path — dispatch to the remaining sinks continues and any producer parked on the incoming queue recovers. ActiveTask's subscriber tap opts in explicitly, so the production wedge is fixed without changing the public primitive's default blocking contract. The consumer of an evicted sink sees the usual QueueShutDown on its next dequeue; no new exception type.

The deeper root cause is subscriber-sink lifecycle — a sink tapped for a consumer that has gone away should be closed at connection teardown rather than discovered full later. That belongs to the request-handler layer and is worth its own issue; the dispatch-layer change above keeps any such sink from taking the task down in the meantime.

I have this implemented with regression tests covering eviction with continued dispatch to other sinks, producer recovery after eviction, preserved backpressure on default taps, and the subscriber tap opting in. The eviction tests go red if the fix is reverted, and the full suite is green with it in place. Can open the PR right away if the direction works for you.