fix: Handle race condition when client disconnects before streaming response completes

When SSE streaming clients disconnect before task completion, background
consumption continues and eventually calls subscriber.onComplete(). By this
time, the HTTP response has already been written and closed by Vert.x,
causing IllegalStateException: "Response has already been written".

This is a benign race condition that occurs when:
1. Client disconnects → HTTP response written/closed
2. Background consumption completes → tries to call subscriber.onComplete()
3. Response already closed → IllegalStateException thrown
4. Error logged by Mutiny: "Mutiny had to drop the following exception"

The fix catches the specific exception message and logs it at debug level,
allowing graceful handling of the expected client disconnect scenario.

Fixes SSE streaming test timeouts on REST transport in CI.

Evidence: Run_4 logs showing ForkJoinPool.commonPool-worker threads throwing
IllegalStateException during onComplete for streaming responses.

Author: kabir

.serena/memories/event-queue-architecture.md

@@ -0,0 +1,224 @@
+# Event Queue Architecture
+
+## Overview
+The A2A Java SDK uses a producer-consumer pattern for handling asynchronous agent execution and event streaming.
+
+## Core Components
+
+### EventQueue Hierarchy
+- **MainQueue**: Primary queue for a task, manages child queues
+- **ChildQueue**: "Tapped" view that receives copies of parent events
+- **Reference Counting**: MainQueue tracks children, closes when all children close (or immediate flag set)
+
+### Queue Lifecycle Management
+
+**Creation:**
+```java
+EventQueue queue = queueManager.createOrTap(taskId);
+// Creates MainQueue if new, or ChildQueue if tapping existing
+```
+
+**Closing:**
+```java
+queue.close(immediate, notifyParent);
+// immediate: clear pending events vs graceful drain
+// notifyParent: decrement MainQueue ref count (ChildQueue only)
+```
+
+### Thread Model
+
+**Agent-Executor Threads:**
+- Pool of 15 threads (configurable via `a2a.executor.max-pool-size`)
+- Execute `AgentExecutor.execute()` which enqueues events
+- Return immediately after execution (as of fix 3e93dd2)
+- No longer block waiting for consumer to start
+
+**Vert.x Worker Threads:**
+- Handle HTTP/gRPC/REST requests
+- Create and run EventConsumer to poll queue
+- Process events via ResultAggregator
+- Close queue on final events/errors
+
+**Background Threads:**
+- Cleanup tasks tracked in `backgroundTasks` set
+- Manage ChildQueue/MainQueue lifecycle after agent completion
+- Handle reference counting for resubscription support
+
+## Event Flow
+
+### Non-Streaming (onMessageSend)
+```
+1. Vert.x worker: Create EventQueue, EventConsumer
+2. Agent-executor: Execute agent async → enqueue events → return immediately
+3. Vert.x worker: Poll queue in consumeAndBreakOnInterrupt()
+   - Blocks until terminal event or AUTH_REQUIRED
+   - Consumer closes queue on final event
+4. Background: Wait for agent → cleanup ChildQueue → manage MainQueue refs
+```
+
+### Streaming (onMessageSendStream)
+```
+1. Vert.x worker: Create EventQueue, EventConsumer
+2. Agent-executor: Execute agent async → enqueue events → return immediately  
+3. Vert.x worker: Return Flow.Publisher immediately
+   - Consumer polls queue in background
+   - Streams events to client
+   - On client disconnect: continue consuming in background
+   - Consumer closes queue on final event
+4. Background: Always async cleanup for streaming
+```
+
+## Queue Closing Semantics
+
+### Who Closes What
+
+**EventConsumer.consumeAll()** - Primary queue closer:
+- Closes queue when final event received (COMPLETED, CANCELED, FAILED, REJECTED, UNKNOWN)
+- Closes queue on Message event (legacy)
+- Closes queue on EventQueueClosedException
+
+**Background cleanupProducer()** - Secondary cleanup:
+- Waits for agent completion
+- Closes ChildQueue (may already be closed by consumer)
+- Uses `close(immediate=false, notifyParent=!keepMainQueueAlive)` for reference counting
+
+**Error Handling:**
+- Agent errors: Set via EnhancedRunnable.setError(), consumer picks up via callback
+- Consumer errors: EventConsumer fails tube, queue closes
+- No immediate queue close in agent whenComplete() callback
+
+### Reference Counting for Resubscription
+
+**Purpose:** Keep MainQueue alive for resubscription when task is non-final
+
+**Mechanism:**
+```java
+boolean keepMainQueueAlive = !isStreaming && task != null && !task.isFinal();
+queue.close(false, !keepMainQueueAlive);
+```
+
+**Effect:**
+- Non-streaming non-final tasks: Close ChildQueue without notifying MainQueue
+- MainQueue stays open, allowing onResubscribeToTask() to tap it
+- Streaming or final tasks: Close ChildQueue and notify MainQueue (decrements ref count)
+
+## Signal Mechanism (Deprecated)
+
+### awaitQueuePollerStart() - NO LONGER USED as of 3e93dd2
+
+**Original Purpose:** 
+Agent-executor threads waited for consumer to start polling before returning, ensuring queue wouldn't close prematurely.
+
+**Problem:**
+Created cascading delays when Vert.x workers were busy. Agent threads blocked up to 10 seconds waiting for worker threads to become available.
+
+**Replacement:**
+Queue lifecycle entirely managed by consumer. Consumer is guaranteed to be running when it closes queue, eliminating race condition.
+
+**Implementation Details:**
+- CountDownLatch in MainQueue, countdown in dequeueEvent() finally block
+- Still exists in code but no longer called
+- Can be removed in future cleanup
+
+## Configuration
+
+### Thread Pool Settings
+```properties
+# tck/src/main/resources/application.properties
+a2a.executor.core-pool-size=5
+a2a.executor.max-pool-size=15
+a2a.executor.keep-alive-seconds=60
+```
+
+### Vert.x Settings  
+```properties
+# Worker thread blocking timeout
+quarkus.vertx.max-worker-execute-time=300s  # Temporary workaround, should be removed
+```
+
+### Queue Settings
+```java
+EventQueue.DEFAULT_QUEUE_SIZE = 1000
+EventConsumer.QUEUE_WAIT_MILLISECONDS = 500  # Polling timeout
+```
+
+## Best Practices
+
+### Creating Queues
+- Use `queueManager.createOrTap()` for both create and resubscribe scenarios
+- MainQueue created with onClose callback for QueueManager cleanup
+- ChildQueues inherit MainQueue events via internalEnqueueEvent()
+
+### Closing Queues  
+- Let EventConsumer.consumeAll() handle queue closing (it knows when to close)
+- Background cleanup manages reference counting
+- Don't close queue in agent execution code
+
+### Resubscription Support
+- Non-streaming non-final tasks keep MainQueue alive
+- Use `queue.close(false, false)` to close ChildQueue without notifying parent
+- MainQueue closes when last child notifies or immediate close requested
+
+### Error Handling
+- Set errors via EnhancedRunnable.setError() for consumer pickup
+- Consumer handles errors via createAgentRunnableDoneCallback()
+- Background cleanup is safe to call even if queue already closed
+
+## Common Patterns
+
+### Standard Request Handler Pattern
+```java
+EventQueue queue = queueManager.createOrTap(taskId);
+EnhancedRunnable producer = registerAndExecuteAgentAsync(taskId, context, queue);
+try {
+    EventConsumer consumer = new EventConsumer(queue);
+    producer.addDoneCallback(consumer.createAgentRunnableDoneCallback());
+    // Consume events (blocking or streaming)
+} finally {
+    runningAgents.remove(taskId);
+    trackBackgroundTask(cleanupProducer(agentFuture, taskId, queue, isStreaming));
+}
+```
+
+### Cancel Pattern  
+```java
+EventQueue queue = queueManager.tap(taskId);  // Tap existing or create new
+agentExecutor.cancel(context, queue);
+Optional.ofNullable(runningAgents.get(taskId)).ifPresent(cf -> cf.cancel(true));
+EventConsumer consumer = new EventConsumer(queue);
+EventKind result = resultAggregator.consumeAll(consumer);
+```
+
+### Resubscribe Pattern
+```java
+EventQueue queue = queueManager.tap(taskId);
+if (queue == null && !task.isFinal()) {
+    queue = queueManager.createOrTap(taskId);  // Create new for future events
+}
+EventConsumer consumer = new EventConsumer(queue);
+return resultAggregator.consumeAndEmit(consumer);
+```
+
+## Troubleshooting
+
+### Queue Not Closing
+- Check if consumer reached final event (DEBUG logs show "Dequeued event")
+- Verify task state is actually final (isFinal() returns true)
+- Check background cleanup completed (CLEANUP END log message)
+
+### Events Missing
+- Verify queue not closed before consumer started polling
+- Check for EventQueueClosedException in logs
+- Ensure agent enqueued events before returning
+
+### Resource Leaks  
+- Monitor runningAgents.size() (should return to 0)
+- Check backgroundTasks.size() (tasks should complete)
+- Verify QueueManager active queue count
+- Use thread dumps to identify blocked threads
+
+### Performance Issues
+- Check for blocking waits in agent-executor threads (should be none)
+- Monitor Vert.x worker thread availability
+- Review queue wait timeouts and polling frequency
+- Consider increasing thread pool size if CPU allows

.serena/memories/tck-timeout-investigation-resolution.md

@@ -0,0 +1,137 @@
+# TCK Timeout Investigation & Resolution
+
+## Problem Summary
+TCK tests were timing out in CI environment after 5 minutes, with 11 leaked agents that never completed cleanup (183 agents registered, only 172 removed from runningAgents map).
+
+## Investigation Timeline
+
+### Initial Hypothesis (Incorrect)
+- Suspected finally blocks not executing due to Vert.x thread abandonment
+- Evidence: "Thread blocked for 60770ms, time limit is 60000ms" in logs
+- Applied fix: Increased Vert.x timeout from 60s to 300s
+- Result: Leak pattern persisted (183 registered, 172 removed, 11 leaked)
+
+### Root Cause Discovery
+Analyzed CI logs and discovered the actual bottleneck:
+
+**Pattern from logs:**
+```
+16:17:43 - Agent waits for queue poller to start
+16:17:53 - Queue poller started (10 seconds later)
+16:18:13 - Next agent waits for queue poller to start  
+16:18:23 - Queue poller started (10 seconds later)
+```
+
+**Root Cause:**
+- Agent-executor threads call `awaitQueuePollerStart()` which blocks up to 10 seconds
+- When Vert.x worker threads are all busy, consumers can't start polling immediately
+- Agent-executor threads (limited pool of 15) block waiting for worker threads
+- Cascading delays: 11 agents × 10 seconds = 110+ seconds of accumulated waste
+- Test timeout at 5 minutes kills process before cleanup finally blocks execute
+
+**Architecture Flaw:**
+Agent-executor threads were synchronously waiting for Vert.x worker threads to become available, creating a resource bottleneck where valuable executor threads sat idle.
+
+## Solution Implemented
+
+### Commit: 3e93dd2
+**Title:** fix: Remove awaitQueuePollerStart to eliminate thread blocking bottleneck
+
+### Key Changes
+
+**1. Removed Blocking Wait**
+File: `server-common/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java`
+
+Before:
+```java
+agentExecutor.execute(requestContext, queue);
+try {
+    queueManager.awaitQueuePollerStart(queue);
+} catch (InterruptedException e) {
+    Thread.currentThread().interrupt();
+}
+```
+
+After:
+```java
+agentExecutor.execute(requestContext, queue);
+// No longer wait for queue poller to start - the consumer (which is guaranteed
+// to be running on the Vert.x worker thread) will handle queue lifecycle.
+// This avoids blocking agent-executor threads waiting for worker threads.
+```
+
+**2. Moved Queue Lifecycle to Consumer**
+
+Before: `whenComplete()` callback closed queue immediately on errors and final states
+
+After: Queue lifecycle entirely managed by `EventConsumer.consumeAll()`:
+- Consumer closes queue on final events (line 95 of EventConsumer.java)
+- Consumer handles errors via error callback mechanism
+- No race condition: consumer is running before it can close queue
+
+**3. Architecture Documentation**
+Added comprehensive JavaDoc explaining new queue lifecycle:
+- Agent-executor thread: Executes and returns immediately
+- Vert.x worker (consumer): Polls queue, closes on final event
+- Background cleanup: Manages ChildQueue/MainQueue reference counting
+
+### Benefits
+1. **Eliminates blocking:** Agent-executor threads return immediately
+2. **No cascading delays:** No more 10-second waits accumulating
+3. **Better resource utilization:** Threads don't block idle
+4. **Simpler architecture:** Consumer owns queue lifecycle (single responsibility)
+5. **No race conditions:** Consumer guaranteed running before closing queue
+
+### Temporary Workaround Status
+File: `tck/src/main/resources/application.properties`
+
+```properties
+# TEMPORARY WORKAROUND: Increase Vert.x worker thread blocking timeout
+# Default is 60s, increased to 300s (5 minutes) to match TCK timeout
+# TODO: Remove this workaround once proper fix is validated:
+#       - Make EventQueue polling non-blocking with async/reactive patterns
+#       - Or add timeout to consumeAndBreakOnInterrupt() to return early
+# See: Investigation of TCK timeout issues in CI (21 agents leaked due to thread abandonment)
+quarkus.vertx.max-worker-execute-time=300s
+```
+
+This workaround prevented Vert.x from abandoning threads, but didn't fix the root cause (blocking wait).
+Should be removed once CI validates the awaitQueuePollerStart fix.
+
+## Expected CI Results
+After this fix, CI should show:
+- ✅ All 183 agents register and all 183 clean up (no leaks)
+- ✅ Tests complete in 2-3 minutes (not 5+ minutes)
+- ✅ No "Waiting for queue poller to start" log entries with long delays
+- ✅ All finally blocks execute successfully
+- ✅ runningAgents map size returns to 0 at test completion
+
+## Technical Details
+
+### EventQueue Lifecycle Before Fix
+1. Agent-executor: Execute agent → enqueue events → **BLOCK** waiting for consumer to start
+2. Vert.x worker: Start consumer → first dequeue signals latch → agent-executor unblocks
+3. Agent-executor whenComplete: Close queue on final state
+4. Consumer: May still be polling when queue closes (race condition prevented by latch)
+
+### EventQueue Lifecycle After Fix  
+1. Agent-executor: Execute agent → enqueue events → **RETURN IMMEDIATELY**
+2. Vert.x worker: Start consumer → poll queue → process events → close on final event
+3. Agent-executor whenComplete: Set error flag, invoke callbacks (no queue operations)
+4. Background cleanup: Wait for agent complete → close ChildQueue → manage MainQueue refs
+
+### Key Architectural Insight
+The consumer (EventConsumer.consumeAll()) is guaranteed to be running when it needs to close the queue, eliminating the need for synchronization between agent-executor threads and Vert.x worker threads.
+
+## Related Files
+- `server-common/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java` - Main fix
+- `server-common/src/main/java/io/a2a/server/events/EventConsumer.java` - Queue lifecycle owner
+- `server-common/src/main/java/io/a2a/server/events/EventQueue.java` - Unused awaitQueuePollerStart() method
+- `tck/src/main/resources/application.properties` - Temporary workaround (to be removed)
+- `.github/workflows/run-tck.yml` - Thread/heap dump capture for diagnostics
+
+## Session Date
+2025-10-13
+
+## Branch
+event-queues-take3