fix: remove redundant awaitingFinalEvent clear and clarify counter logic

The awaitingFinalEvent flag is already cleared in internalEnqueueItem
when the final event arrives in the MainQueue. The dequeue-side clearing
in ChildQueue.dequeueEventItem was a redundant no-op on the normal path
and has been removed to avoid misleading future maintainers.

Also clarifies the pollTimeoutsWhileAwaitingFinal reset logic in
EventConsumer by documenting all three cases of the if/else-if block:
- awaitingFinal && queueSize == 0: increment counter, give up at MAX
- awaitingFinal && queueSize > 0: implicit fall-through, do nothing
- !awaitingFinal: reset counter (distinct from the successful-dequeue reset)

Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>

Author: ehsavoie

server-common/src/main/java/io/a2a/server/events/EventConsumer.java

@@ -31,12 +31,14 @@ public class EventConsumer {
     private static final int QUEUE_WAIT_MILLISECONDS = 500;
     // In replicated scenarios, events can arrive hundreds of milliseconds after local agent completes
     // Grace period allows Kafka replication to deliver late-arriving events
-    // 3 timeouts * 500ms = 1500ms grace period for replication delays
+    // Calculation: MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED * QUEUE_WAIT_MILLISECONDS = 1500ms
     private static final int MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED = 3;
     // Maximum time to wait for final event when awaitingFinalEvent is set
     // If event doesn't arrive after this many timeouts, assume it won't arrive
-    // 6 timeouts * 500ms = 3000ms maximum wait for final event arrival
-    private static final int MAX_POLL_TIMEOUTS_AWAITING_FINAL = 6;
+    // Calculation: MAX_POLL_TIMEOUTS_AWAITING_FINAL * QUEUE_WAIT_MILLISECONDS = 3000ms
+    private static final int MAX_AWAITING_FINAL_TIMEOUT_MS = 3000;
+    private static final int MAX_POLL_TIMEOUTS_AWAITING_FINAL = 
+        MAX_AWAITING_FINAL_TIMEOUT_MS / QUEUE_WAIT_MILLISECONDS;
 
     public EventConsumer(EventQueue queue) {
         this.queue = queue;
@@ -108,7 +110,14 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
                             // proceed with normal timeout logic to prevent infinite waiting.
                             boolean isInterruptedState = lastSeenTaskState != null && lastSeenTaskState.isInterrupted();
 
-                            // Track how long we've been waiting for the final event
+                            // Track how long we've been waiting for the final event.
+                            // Three cases for the awaiting counter:
+                            //   awaitingFinal && queueSize == 0: final event enqueued in MainQueue but not yet
+                            //     distributed here — increment timeout counter and give up after MAX timeout.
+                            //   awaitingFinal && queueSize > 0: events are still in transit, do nothing —
+                            //     the counter is reset below once an event is successfully dequeued.
+                            //   !awaitingFinal: not waiting for anything — reset the counter (timeout case;
+                            //     the successful-dequeue reset happens below at the event-received path).
                             if (awaitingFinal && queueSize == 0) {
                                 pollTimeoutsWhileAwaitingFinal++;
                                 if (pollTimeoutsWhileAwaitingFinal >= MAX_POLL_TIMEOUTS_AWAITING_FINAL) {
@@ -118,8 +127,10 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
                                     queue.clearAwaitingFinalEvent();
                                     awaitingFinal = false; // Also update local variable for this iteration
                                 }
-                            } else {
-                                pollTimeoutsWhileAwaitingFinal = 0; // Reset when event arrives or queue not awaiting
+                            } else if (!awaitingFinal) {
+                                // Poll timed out and we are not awaiting a final event: reset the counter.
+                                // (The successful-dequeue reset is handled separately below.)
+                                pollTimeoutsWhileAwaitingFinal = 0;
                             }
 
                             if (agentCompleted && queueSize == 0 && !isInterruptedState && !awaitingFinal) {