Gemini review

Author: kabir

.claude/architecture/eventqueue/FLOWS.md

@@ -164,21 +164,26 @@ Bridges EventConsumer and DefaultRequestHandler with three consumption modes:
 
 ## Cleanup Integration
 
-### Non-Streaming Cleanup Decision
+### Actual Implementation: Always Asynchronous
+
+**Reality**: Cleanup is ALWAYS asynchronous in both streaming and non-streaming flows. The cleanup happens in the `finally` block via `cleanupProducer()`, which runs in a background thread.
 
 ```java
-if (event instanceof Message || isFinalEvent(event)) {
-    if (!interrupted) {
-        cleanup(queue, task, false);  // Immediate: wait for agent, close queue
-    } else {
-        cleanup(queue, task, true);   // Async: close in background (AUTH_REQUIRED case)
-    }
-}
+// Both flows (in finally block):
+cleanupProducer(agentFuture, consumptionFuture, taskId, queue, isStreaming)
+    .whenComplete((res, err) -> {
+        if (err != null) {
+            LOGGER.error("Error during async cleanup for task {}", taskId, err);
+        }
+    });
 ```
 
-**Logic**:
-- Terminal event + not interrupted → Immediate cleanup (wait for agent, close queue)
-- Terminal event + interrupted (AUTH_REQUIRED) → Async cleanup (agent still running)
+**Key Points**:
+- Cleanup is initiated in `finally` block regardless of flow outcome
+- `cleanupProducer()` waits for both agent and consumption futures to complete
+- Queue closure happens in background, never blocking the request thread
+- For streaming: EventConsumer manages queue lifecycle via `agentCompleted` flag
+- For non-streaming: Queue is closed directly after agent completes
 
 ### Streaming Cleanup
 

.claude/architecture/eventqueue/LIFECYCLE.md

@@ -130,17 +130,32 @@ cleanup(queue, task, true);  // Background cleanup after streaming completes
 ### Cleanup Implementation
 
 ```java
-private void cleanup(EventQueue queue, Task task, boolean async) {
-    Runnable cleanupTask = () -> {
-        agentFuture.join();  // Wait for agent to finish
-        queue.close();       // Close ChildQueue (triggers Level 2 check)
-    };
-
-    if (async) {
-        CompletableFuture.runAsync(cleanupTask, executor);
-    } else {
-        cleanupTask.run();
+private CompletableFuture<Void> cleanupProducer(
+        @Nullable CompletableFuture<Void> agentFuture,
+        @Nullable CompletableFuture<Void> consumptionFuture,
+        String taskId,
+        EventQueue queue,
+        boolean isStreaming) {
+
+    if (agentFuture == null) {
+        return CompletableFuture.completedFuture(null);
+    }
+
+    // Wait for BOTH agent AND consumption to complete before cleanup
+    CompletableFuture<Void> bothComplete = agentFuture;
+    if (consumptionFuture != null) {
+        bothComplete = CompletableFuture.allOf(agentFuture, consumptionFuture);
     }
+
+    return bothComplete.whenComplete((v, t) -> {
+        if (isStreaming) {
+            // EventConsumer manages queue lifecycle via agentCompleted flag
+            LOGGER.debug("Streaming: queue lifecycle managed by EventConsumer");
+        } else {
+            // Close ChildQueue directly (triggers Level 2 check)
+            queue.close(false, true);
+        }
+    });
 }
 ```
 

.claude/architecture/eventqueue/SCENARIOS.md

@@ -105,13 +105,19 @@ EventQueue mainQueue = queueManager.createOrTap(taskId);
 // Second client taps into existing MainQueue
 EventQueue childQueue = queueManager.tap(taskId);
 
-// Event distribution (MainQueue.enqueueEvent)
+// Event distribution (ASYNCHRONOUS via MainEventBus)
+// NOTE: Distribution is NOT immediate!
 public void enqueueEvent(Event event) {
-    super.enqueueEvent(event);  // Enqueue to MainEventBus
-    children.forEach(child -> child.internalEnqueueEvent(event));  // Copy to ChildQueues
-    if (enqueueHook != null) {
-        enqueueHook.onEnqueue(event);  // Replication hook
-    }
+    // Step 1: Submit to MainEventBus (async processing)
+    mainEventBus.submit(event);
+
+    // Step 2: MainEventBusProcessor thread (separate background thread):
+    //   - Persists event to TaskStore
+    //   - Distributes to all ChildQueues via child.internalEnqueueItem(item)
+    //   - Invokes replication hook if configured
+
+    // Key Point: Events are NOT immediately in ChildQueues!
+    // There's a delay while MainEventBusProcessor persists and distributes.
 }
 ```