fix: Implement reference counting for EventQueue to prevent premature MainQueue closure

Previously, MainQueues closed immediately when any ChildQueue closed, causing
race conditions where resubscription attempts would fail with TaskNotFoundError.
This was particularly problematic in scenarios with multiple concurrent consumers
or when clients resubscribed to active tasks.

This commit introduces a reference counting mechanism where MainQueues track
active ChildQueues and only close when all children have closed, preventing
premature closure while consumers are still active.

Changes:

EventQueue architecture:
- MainQueue now maintains a list of active ChildQueues
- ChildQueue.close() notifies parent via childClosing() callback
- MainQueue only closes when: immediate=true OR all children closed
- Added getActiveChildCount() for testing reference counting mechanism

Non-blocking message support:
- Non-blocking non-final tasks keep MainQueue alive for resubscription
- Added close(immediate, notifyParent) method to EventQueue
- ChildQueue can close without decrementing parent reference count
- DefaultRequestHandler detects non-blocking non-final scenarios
- Support multiple messages to same task after non-blocking sendMessage
- AUTH_REQUIRED state now continues processing in background

ResultAggregator improvements:
- consumeAndBreakOnInterrupt() supports blocking and non-blocking modes
- AUTH_REQUIRED events trigger background continuation for additional messages
- Refactored to use AsyncUtils.consumer() for consistent event processing
- EventConsumer.close() explicitly closes queue to prevent resource leaks

Robustness improvements:
- onResubscribeToTask() calls createOrTap() for non-final tasks when queue missing
- Provides recovery path for edge cases (server restart, explicit queue closure)
- Allows client resubscription even if MainQueue was unexpectedly removed
- Note: historical events unavailable, only future events delivered

Test coverage:
- Added testMainQueueReferenceCountingWithMultipleConsumers()
- Added testNonBlockingWithMultipleMessages()
- Verified TCK compliance for resubscription behavior

Fixes race condition where:
1. Client A subscribes to task → ChildQueue A created
2. Client B resubscribes → ChildQueue B created
3. Client A disconnects → ChildQueue A closes → MainQueue closed prematurely
4. Client B receives no further events (incorrect)

With reference counting:
- MainQueue stays alive while any ChildQueue exists
- All active consumers receive events until MainQueue closes
- Resubscription works reliably for active tasks
- Multiple messages can be sent to same non-blocking task

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: kabir

.serena/.gitignore

@@ -0,0 +1 @@
+/cache

.serena/cache/java/document_symbols_cache_v23-06-25.pkl

@@ -0,0 +1,67 @@
+# language of the project (csharp, python, rust, java, typescript, go, cpp, or ruby)
+#  * For C, use cpp
+#  * For JavaScript, use typescript
+# Special requirements:
+#  * csharp: Requires the presence of a .sln file in the project folder.
+language: java
+
+# whether to use the project's gitignore file to ignore files
+# Added on 2025-04-07
+ignore_all_files_in_gitignore: true
+# list of additional paths to ignore
+# same syntax as gitignore, so you can use * and **
+# Was previously called `ignored_dirs`, please update your config if you are using that.
+# Added (renamed) on 2025-04-07
+ignored_paths: []
+
+# whether the project is in read-only mode
+# If set to true, all editing tools will be disabled and attempts to use them will result in an error
+# Added on 2025-04-18
+read_only: false
+
+# list of tool names to exclude. We recommend not excluding any tools, see the readme for more details.
+# Below is the complete list of tools for convenience.
+# To make sure you have the latest list of tools, and to view their descriptions, 
+# execute `uv run scripts/print_tool_overview.py`.
+#
+#  * `activate_project`: Activates a project by name.
+#  * `check_onboarding_performed`: Checks whether project onboarding was already performed.
+#  * `create_text_file`: Creates/overwrites a file in the project directory.
+#  * `delete_lines`: Deletes a range of lines within a file.
+#  * `delete_memory`: Deletes a memory from Serena's project-specific memory store.
+#  * `execute_shell_command`: Executes a shell command.
+#  * `find_referencing_code_snippets`: Finds code snippets in which the symbol at the given location is referenced.
+#  * `find_referencing_symbols`: Finds symbols that reference the symbol at the given location (optionally filtered by type).
+#  * `find_symbol`: Performs a global (or local) search for symbols with/containing a given name/substring (optionally filtered by type).
+#  * `get_current_config`: Prints the current configuration of the agent, including the active and available projects, tools, contexts, and modes.
+#  * `get_symbols_overview`: Gets an overview of the top-level symbols defined in a given file.
+#  * `initial_instructions`: Gets the initial instructions for the current project.
+#     Should only be used in settings where the system prompt cannot be set,
+#     e.g. in clients you have no control over, like Claude Desktop.
+#  * `insert_after_symbol`: Inserts content after the end of the definition of a given symbol.
+#  * `insert_at_line`: Inserts content at a given line in a file.
+#  * `insert_before_symbol`: Inserts content before the beginning of the definition of a given symbol.
+#  * `list_dir`: Lists files and directories in the given directory (optionally with recursion).
+#  * `list_memories`: Lists memories in Serena's project-specific memory store.
+#  * `onboarding`: Performs onboarding (identifying the project structure and essential tasks, e.g. for testing or building).
+#  * `prepare_for_new_conversation`: Provides instructions for preparing for a new conversation (in order to continue with the necessary context).
+#  * `read_file`: Reads a file within the project directory.
+#  * `read_memory`: Reads the memory with the given name from Serena's project-specific memory store.
+#  * `remove_project`: Removes a project from the Serena configuration.
+#  * `replace_lines`: Replaces a range of lines within a file with new content.
+#  * `replace_symbol_body`: Replaces the full definition of a symbol.
+#  * `restart_language_server`: Restarts the language server, may be necessary when edits not through Serena happen.
+#  * `search_for_pattern`: Performs a search for a pattern in the project.
+#  * `summarize_changes`: Provides instructions for summarizing the changes made to the codebase.
+#  * `switch_modes`: Activates modes by providing a list of their names
+#  * `think_about_collected_information`: Thinking tool for pondering the completeness of collected information.
+#  * `think_about_task_adherence`: Thinking tool for determining whether the agent is still on track with the current task.
+#  * `think_about_whether_you_are_done`: Thinking tool for determining whether the task is truly completed.
+#  * `write_memory`: Writes a named memory (for future reference) to Serena's project-specific memory store.
+excluded_tools: []
+
+# initial prompt for the project. It will always be given to the LLM upon activating the project
+# (contrary to the memories, which are loaded on demand).
+initial_prompt: ""
+
+project_name: "a2a-java"

.serena/project.yml

@@ -94,6 +94,11 @@ public EventQueue.EventQueueBuilder getEventQueueBuilder(String taskId) {
                 .hook(new ReplicationHook(taskId));
     }
 
+    @Override
+    public int getActiveChildQueueCount(String taskId) {
+        return delegate.getActiveChildQueueCount(taskId);
+    }
+
     private class ReplicatingEventQueueFactory implements EventQueueFactory {
         @Override
         public EventQueue.EventQueueBuilder builder(String taskId) {

extras/queue-manager-replicated/core/src/main/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManager.java

@@ -6,6 +6,7 @@
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -17,6 +18,7 @@
 
 import io.a2a.server.events.EventQueue;
 import io.a2a.server.events.EventQueueClosedException;
+import io.a2a.server.events.EventQueueTestHelper;
 import io.a2a.spec.Event;
 import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.TaskState;
@@ -147,8 +149,11 @@ void testBasicQueueManagerFunctionality() throws InterruptedException {
         EventQueue queue = queueManager.createOrTap(taskId);
         assertNotNull(queue);
 
+        // createOrTap now returns ChildQueue, get returns MainQueue
         EventQueue retrievedQueue = queueManager.get(taskId);
-        assertEquals(queue, retrievedQueue);
+        assertNotNull(retrievedQueue);
+        // queue should be a ChildQueue (cannot be tapped)
+        assertThrows(IllegalStateException.class, () -> EventQueueTestHelper.tapQueue(queue));
 
         EventQueue tappedQueue = queueManager.tap(taskId);
         assertNotNull(tappedQueue);

extras/queue-manager-replicated/core/src/test/java/io/a2a/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java

@@ -0,0 +1,10 @@
+package io.a2a.server.events;
+
+/**
+ * Utils to access package private methods in the io.a2a.server.events package
+ */
+public class EventQueueTestHelper {
+    public static EventQueue tapQueue(EventQueue queue) {
+        return queue.tap();
+    }
+}

extras/queue-manager-replicated/core/src/test/java/io/a2a/server/events/EventQueueTestHelper.java

@@ -105,6 +105,14 @@ public Response getStreamingSubscribedCount() {
         return Response.ok(String.valueOf(streamingSubscribedCount.get()), TEXT_PLAIN).build();
     }
 
+    @GET
+    @Path("/queue/childCount/{taskId}")
+    @Produces(TEXT_PLAIN)
+    public Response getChildQueueCount(@PathParam("taskId") String taskId) {
+        int count = testUtilsBean.getChildQueueCount(taskId);
+        return Response.ok(String.valueOf(count), TEXT_PLAIN).build();
+    }
+
     @DELETE
     @Path("/task/{taskId}/config/{configId}")
     public Response deleteTaskPushNotificationConfig(@PathParam("taskId") String taskId, @PathParam("configId") String configId) {

reference/grpc/src/test/java/io/a2a/server/grpc/quarkus/A2ATestResource.java

@@ -139,6 +139,14 @@ public void getStreamingSubscribedCount(RoutingContext rc) {
                 .end(String.valueOf(streamingSubscribedCount.get()));
     }
 
+    @Route(path = "/test/queue/childCount/:taskId", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+    public void getChildQueueCount(@Param String taskId, RoutingContext rc) {
+        int count = testUtilsBean.getChildQueueCount(taskId);
+        rc.response()
+                .setStatusCode(200)
+                .end(String.valueOf(count));
+    }
+
     @Route(path = "/test/task/:taskId/config/:configId", methods = {Route.HttpMethod.DELETE}, type = Route.HandlerType.BLOCKING)
     public void deleteTaskPushNotificationConfig(@Param String taskId, @Param String configId, RoutingContext rc) {
         try {

reference/jsonrpc/src/test/java/io/a2a/server/apps/quarkus/A2ATestRoutes.java

@@ -1 +1 @@
-quarkus.arc.selected-alternatives=io.a2a.server.apps.common.TestHttpClient
+quarkus.arc.selected-alternatives=io.a2a.server.apps.common.TestHttpClient

reference/jsonrpc/src/test/resources/application.properties

@@ -141,6 +141,14 @@ public void getStreamingSubscribedCount(RoutingContext rc) {
                 .end(String.valueOf(streamingSubscribedCount.get()));
     }
 
+    @Route(path = "/test/queue/childCount/:taskId", methods = {Route.HttpMethod.GET}, produces = {TEXT_PLAIN})
+    public void getChildQueueCount(@Param String taskId, RoutingContext rc) {
+        int count = testUtilsBean.getChildQueueCount(taskId);
+        rc.response()
+                .setStatusCode(200)
+                .end(String.valueOf(count));
+    }
+
     @Route(path = "/test/task/:taskId/config/:configId", methods = {Route.HttpMethod.DELETE}, type = Route.HandlerType.BLOCKING)
     public void deleteTaskPushNotificationConfig(@Param String taskId, @Param String configId, RoutingContext rc) {
         try {