Debug logging

kabir · kabir · commit 54c02c83dda2 · 2026-01-07T15:54:56.000Z
diff --git a/server-common/src/main/java/io/a2a/server/events/EventConsumer.java b/server-common/src/main/java/io/a2a/server/events/EventConsumer.java
@@ -59,11 +59,15 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
                     EventQueueItem item;
                     Event event;
                     try {
+                        LOGGER.debug("EventConsumer polling queue {} (error={})", System.identityHashCode(queue), error);
                         item = queue.dequeueEventItem(QUEUE_WAIT_MILLISECONDS);
                         if (item == null) {
+                            LOGGER.debug("EventConsumer poll timeout (null item), continuing");
                             continue;
                         }
                         event = item.getEvent();
+                        LOGGER.debug("EventConsumer received event: {} (queue={})",
+                            event.getClass().getSimpleName(), System.identityHashCode(queue));
 
                         // Defensive logging for error handling
                         if (event instanceof Throwable thr) {
@@ -125,8 +129,13 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
 
     public EnhancedRunnable.DoneCallback createAgentRunnableDoneCallback() {
         return agentRunnable -> {
+            LOGGER.info("EventConsumer: Agent done callback invoked (hasError={}, queue={})",
+                agentRunnable.getError() != null, System.identityHashCode(queue));
             if (agentRunnable.getError() != null) {
                 error = agentRunnable.getError();
+                LOGGER.info("EventConsumer: Set error field from agent callback");
+            } else {
+                LOGGER.info("EventConsumer: Agent completed successfully (no error), continuing consumption");
             }
         };
     }
diff --git a/server-common/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java
@@ -374,6 +374,17 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
 
         boolean blocking = params.configuration() != null && Boolean.TRUE.equals(params.configuration().blocking());
 
+        // Log blocking behavior from client request
+        if (params.configuration() != null && params.configuration().blocking() != null) {
+            LOGGER.info("DefaultRequestHandler: Client requested blocking={} for task {}",
+                params.configuration().blocking(), taskId);
+        } else if (params.configuration() != null) {
+            LOGGER.info("DefaultRequestHandler: Client sent configuration but blocking=null, using default blocking=true for task {}", taskId);
+        } else {
+            LOGGER.info("DefaultRequestHandler: Client sent no configuration, using default blocking=true for task {}", taskId);
+        }
+        LOGGER.info("DefaultRequestHandler: Final blocking decision: {} for task {}", blocking, taskId);
+
         boolean interruptedOrNonBlocking = false;
 
         EnhancedRunnable producerRunnable = registerAndExecuteAgentAsync(taskId, mss.requestContext, queue);
@@ -395,7 +406,8 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                 throw new InternalError("No result");
             }
             interruptedOrNonBlocking = etai.interrupted();
-            LOGGER.debug("Was interrupted or non-blocking: {}", interruptedOrNonBlocking);
+            LOGGER.info("DefaultRequestHandler: interruptedOrNonBlocking={} (blocking={}, eventType={})",
+                interruptedOrNonBlocking, blocking, kind != null ? kind.getClass().getSimpleName() : null);
 
             // For blocking calls that were interrupted (returned on first event),
             // wait for agent execution and event processing BEFORE returning to client.
@@ -419,29 +431,31 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                 // 2. Close the queue to signal consumption can complete
                 // 3. Wait for consumption to finish processing events
                 // 4. Fetch final task state from TaskStore
+                LOGGER.debug("DefaultRequestHandler: Entering blocking fire-and-forget handling for task {}", taskId);
 
                 try {
                     // Step 1: Wait for agent to finish (with configurable timeout)
                     if (agentFuture != null) {
                         try {
                             agentFuture.get(agentCompletionTimeoutSeconds, SECONDS);
-                            LOGGER.debug("Agent completed for task {}", taskId);
+                            LOGGER.debug("DefaultRequestHandler: Step 1 - Agent completed for task {}", taskId);
                         } catch (java.util.concurrent.TimeoutException e) {
                             // Agent still running after timeout - that's fine, events already being processed
-                            LOGGER.debug("Agent still running for task {} after {}s", taskId, agentCompletionTimeoutSeconds);
+                            LOGGER.debug("DefaultRequestHandler: Step 1 - Agent still running for task {} after {}s timeout",
+                                taskId, agentCompletionTimeoutSeconds);
                         }
                     }
 
                     // Step 2: Close the queue to signal consumption can complete
                     // For fire-and-forget tasks, there's no final event, so we need to close the queue
                     // This allows EventConsumer.consumeAll() to exit
                     queue.close(false, false);  // graceful close, don't notify parent yet
-                    LOGGER.debug("Closed queue for task {} to allow consumption completion", taskId);
+                    LOGGER.debug("DefaultRequestHandler: Step 2 - Closed queue for task {} to allow consumption completion", taskId);
 
                     // Step 3: Wait for consumption to complete (now that queue is closed)
                     if (etai.consumptionFuture() != null) {
                         etai.consumptionFuture().get(consumptionCompletionTimeoutSeconds, SECONDS);
-                        LOGGER.debug("Consumption completed for task {}", taskId);
+                        LOGGER.debug("DefaultRequestHandler: Step 3 - Consumption completed for task {}", taskId);
                     }
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
@@ -464,11 +478,11 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                 Task updatedTask = taskStore.get(nonNullTaskId);
                 if (updatedTask != null) {
                     kind = updatedTask;
-                    if (LOGGER.isDebugEnabled()) {
-                        LOGGER.debug("Fetched final task for {} with state {} and {} artifacts",
-                            nonNullTaskId, updatedTask.status().state(),
-                            updatedTask.artifacts().size());
-                    }
+                    LOGGER.debug("DefaultRequestHandler: Step 4 - Fetched final task for {} with state {} and {} artifacts",
+                        taskId, updatedTask.status().state(),
+                        updatedTask.artifacts().size());
+                } else {
+                    LOGGER.warn("DefaultRequestHandler: Step 4 - Task {} not found in TaskStore!", taskId);
                 }
             }
             if (kind instanceof Task taskResult && !taskId.equals(taskResult.id())) {
diff --git a/server-common/src/main/java/io/a2a/server/tasks/ResultAggregator.java b/server-common/src/main/java/io/a2a/server/tasks/ResultAggregator.java
@@ -152,6 +152,9 @@ public EventTypeAndInterrupt consumeAndBreakOnInterrupt(EventConsumer consumer,
                     boolean isAuthRequired = (event instanceof Task task && task.status().state() == TaskState.AUTH_REQUIRED)
                             || (event instanceof TaskStatusUpdateEvent tsue && tsue.status().state() == TaskState.AUTH_REQUIRED);
 
+                    LOGGER.info("ResultAggregator: Evaluating interrupt (blocking={}, isFinal={}, isAuth={}, eventType={})",
+                        blocking, isFinalEvent, isAuthRequired, event.getClass().getSimpleName());
+
                     // Always interrupt on auth_required, as it needs external action.
                     if (isAuthRequired) {
                         // auth-required is a special state: the message should be
@@ -161,23 +164,27 @@ public EventTypeAndInterrupt consumeAndBreakOnInterrupt(EventConsumer consumer,
                         // new request is expected in order for the agent to make progress,
                         // so the agent should exit.
                         shouldInterrupt = true;
+                        LOGGER.info("ResultAggregator: Setting shouldInterrupt=true (AUTH_REQUIRED)");
                     }
                     else if (!blocking) {
                         // For non-blocking calls, interrupt as soon as a task is available.
                         shouldInterrupt = true;
+                        LOGGER.info("ResultAggregator: Setting shouldInterrupt=true (non-blocking)");
                     }
                     else if (blocking) {
                         // For blocking calls: Interrupt to free Vert.x thread, but continue in background
                         // Python's async consumption doesn't block threads, but Java's does
                         // So we interrupt to return quickly, then rely on background consumption
                         shouldInterrupt = true;
+                        LOGGER.info("ResultAggregator: Setting shouldInterrupt=true (blocking, isFinal={})", isFinalEvent);
                         if (LOGGER.isDebugEnabled()) {
                             LOGGER.debug("Blocking call for task {}: {} event, returning with background consumption",
                                 taskIdForLogging(), isFinalEvent ? "final" : "non-final");
                         }
                     }
 
                     if (shouldInterrupt) {
+                        LOGGER.info("ResultAggregator: Interrupting consumption (setting interrupted=true)");
                         // Complete the future to unblock the main thread
                         interrupted.set(true);
                         completionFuture.complete(null);
