fix(server): make reconciliation timeout configurable, add logging

docs/content/configuration.md

@@ -40,6 +40,9 @@ a2a.blocking.agent.timeout.seconds=30
 
 # Timeout for event consumption in blocking calls (default: 5 seconds)
 a2a.blocking.consumption.timeout.seconds=5
+
+# Timeout for TaskStore reconciliation polling in blocking calls (default: 1 second)
+a2a.blocking.reconciliation.timeout.seconds=1
 ```
 
 ### Tuning Guidelines
@@ -48,6 +51,7 @@ a2a.blocking.consumption.timeout.seconds=5
 - **Resource Management**: The dedicated executor prevents streaming operations from competing with the ForkJoinPool.
 - **Concurrency**: In production with high concurrent streaming, increase pool sizes accordingly.
 - **Agent Timeouts**: LLM-based agents may need longer timeouts (60-120s) compared to simple agents.
+- **Reconciliation Timeout**: Increase if blocking calls fail with "Could not find a Task/Message" under heavy load or with slow TaskStore implementations.
 
 ## MicroProfile Config Integration
 

integrations/microprofile-config/README.md

@@ -37,6 +37,7 @@ a2a.executor.max-pool-size=100
 # Timeout configuration
 a2a.blocking.agent.timeout.seconds=60
 a2a.blocking.consumption.timeout.seconds=10
+a2a.blocking.reconciliation.timeout.seconds=2
 ```
 
 **Environment variables:**

server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java

@@ -142,8 +142,9 @@
  *   <li><b>Blocking (configuration.blocking=true):</b> Client waits for first event or final task state</li>
  *   <li><b>Streaming:</b> Client receives events as they arrive via reactive streams</li>
  *   <li>Both modes support fire-and-forget (agent continues after client disconnect)</li>
- *   <li>Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds} and
- *       {@code a2a.blocking.consumption.timeout.seconds}</li>
+ *   <li>Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
+ *       {@code a2a.blocking.consumption.timeout.seconds}, and
+ *       {@code a2a.blocking.reconciliation.timeout.seconds}</li>
  * </ul>
  *
  * <h2>CDI Dependencies</h2>
@@ -189,6 +190,7 @@ public class DefaultRequestHandler implements RequestHandler {
 
     private static final String A2A_BLOCKING_AGENT_TIMEOUT_SECONDS = "a2a.blocking.agent.timeout.seconds";
     private static final String A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS = "a2a.blocking.consumption.timeout.seconds";
+    private static final String A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS = "a2a.blocking.reconciliation.timeout.seconds";
 
     @Inject
     A2AConfigProvider configProvider;
@@ -215,6 +217,19 @@ public class DefaultRequestHandler implements RequestHandler {
      */
     int consumptionCompletionTimeoutSeconds;
 
+    /**
+     * Timeout in seconds for TaskStore reconciliation polling in blocking calls.
+     * When the in-memory event capture is empty, the aggregator polls TaskStore
+     * with a bounded timeout to handle the race where MainEventBusProcessor
+     * has not yet persisted the task.
+     * <p>
+     * Property: {@code a2a.blocking.reconciliation.timeout.seconds}<br>
+     * Default: 1 second<br>
+     * Note: Property override requires a configurable {@link A2AConfigProvider} on the classpath
+     * (e.g., MicroProfileConfigProvider in reference implementations).
+     */
+    int reconciliationTimeoutSeconds;
+
     // Fields set by constructor injection cannot be final. We need a noargs constructor for
     // Jakarta compatibility, and it seems that making fields set by constructor injection
     // final, is not proxyable in all runtimes
@@ -276,6 +291,8 @@ void initConfig() {
                 configProvider.getValue(A2A_BLOCKING_AGENT_TIMEOUT_SECONDS));
         consumptionCompletionTimeoutSeconds = Integer.parseInt(
                 configProvider.getValue(A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS));
+        reconciliationTimeoutSeconds = Integer.parseInt(
+                configProvider.getValue(A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS));
     }
 
 
@@ -291,6 +308,7 @@ public static DefaultRequestHandler create(AgentExecutor agentExecutor, TaskStor
                         mainEventBusProcessor, executor, eventConsumerExecutor);
         handler.agentCompletionTimeoutSeconds = 5;
         handler.consumptionCompletionTimeoutSeconds = 2;
+        handler.reconciliationTimeoutSeconds = 1;
 
         return handler;
     }
@@ -377,7 +395,8 @@ public Task onCancelTask(CancelTaskParams params, ServerCallContext context) thr
                 taskStore,
                 null);
 
-        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor);
+        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor,
+                        SECONDS.toNanos(reconciliationTimeoutSeconds));
 
         EventQueue queue = queueManager.createOrTap(task.id());
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
@@ -450,7 +469,8 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
         // Create queue with real taskId (no tempId parameter needed)
         EventQueue queue = queueManager.createOrTap(queueTaskId);
         final java.util.concurrent.atomic.AtomicReference<@NonNull String> taskId = new java.util.concurrent.atomic.AtomicReference<>(queueTaskId);
-        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor);
+        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor,
+                        SECONDS.toNanos(reconciliationTimeoutSeconds));
 
         // Default to blocking per A2A spec (returnImmediately defaults to false, meaning wait for completion)
         boolean returnImmediately = params.configuration() != null && Boolean.TRUE.equals(params.configuration().returnImmediately());
@@ -668,7 +688,8 @@ public Flow.Publisher<StreamingEventKind> onMessageSendStream(
                     .taskId(taskId.get()).build(), version);
         }
 
-        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor);
+        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor, eventConsumerExecutor,
+                        SECONDS.toNanos(reconciliationTimeoutSeconds));
 
         // Create consumer BEFORE starting agent - callback is registered inside registerAndExecuteAgentAsync
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
@@ -857,7 +878,8 @@ public Flow.Publisher<StreamingEventKind> onSubscribeToTask(TaskIdParams params,
         }
 
         TaskManager taskManager = new TaskManager(task.id(), task.contextId(), taskStore, null);
-        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor);
+        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor, eventConsumerExecutor,
+                        SECONDS.toNanos(reconciliationTimeoutSeconds));
         EventQueue queue = queueManager.tap(task.id());
         LOGGER.debug("onSubscribeToTask - tapped queue: {}", queue != null ? System.identityHashCode(queue) : "null");
 

server-common/src/main/java/org/a2aproject/sdk/server/tasks/ResultAggregator.java

@@ -29,19 +29,21 @@
 
 public class ResultAggregator {
     private static final Logger LOGGER = LoggerFactory.getLogger(ResultAggregator.class);
-    private static final long TASK_STORE_RECONCILIATION_TIMEOUT_NANOS = TimeUnit.SECONDS.toNanos(1);
     private static final long TASK_STORE_RECONCILIATION_POLL_MILLIS = 10;
 
     private final TaskManager taskManager;
     private final Executor executor;
     private final Executor eventConsumerExecutor;
+    private final long reconciliationTimeoutNanos;
     private volatile @Nullable Message message;
 
-    public ResultAggregator(TaskManager taskManager, @Nullable Message message, Executor executor, Executor eventConsumerExecutor) {
+    public ResultAggregator(TaskManager taskManager, @Nullable Message message, Executor executor,
+            Executor eventConsumerExecutor, long reconciliationTimeoutNanos) {
         this.taskManager = taskManager;
         this.message = message;
         this.executor = executor;
         this.eventConsumerExecutor = eventConsumerExecutor;
+        this.reconciliationTimeoutNanos = reconciliationTimeoutNanos;
     }
 
     public @Nullable EventKind getCurrentResult() {
@@ -238,7 +240,8 @@ else if (blocking) {
             Utils.rethrow(error);
         }
 
-        // Return Message if captured, otherwise Task if captured, otherwise reconcile with TaskStore.
+        // Return Message if captured, otherwise Task if captured,
+        // otherwise poll TaskStore with bounded timeout (blocking) or single read (non-blocking).
         EventKind eventKind = message.get();
         if (eventKind == null) {
             eventKind = capturedTask.get();
@@ -268,7 +271,10 @@ else if (blocking) {
             return task;
         }
 
-        long deadline = System.nanoTime() + TASK_STORE_RECONCILIATION_TIMEOUT_NANOS;
+        LOGGER.debug("TaskStore reconciliation: task not found on first read, polling for up to {}ms",
+                TimeUnit.NANOSECONDS.toMillis(reconciliationTimeoutNanos));
+
+        long deadline = System.nanoTime() + reconciliationTimeoutNanos;
         while (System.nanoTime() < deadline) {
             try {
                 Thread.sleep(TASK_STORE_RECONCILIATION_POLL_MILLIS);
@@ -279,6 +285,7 @@ else if (blocking) {
 
             task = taskManager.getTask();
             if (task != null) {
+                LOGGER.debug("TaskStore reconciliation: task {} found after polling", task.id());
                 return task;
             }
         }

server-common/src/main/resources/META-INF/a2a-defaults.properties

@@ -10,6 +10,10 @@ a2a.blocking.agent.timeout.seconds=30
 # Ensures TaskStore is fully updated before returning to client
 a2a.blocking.consumption.timeout.seconds=5
 
+# Timeout for TaskStore reconciliation polling in blocking calls (seconds)
+# When in-memory event capture is empty, polls TaskStore with this bounded timeout
+a2a.blocking.reconciliation.timeout.seconds=1
+
 # AsyncExecutorProducer - Thread pool configuration
 # Core pool size for async agent execution
 a2a.executor.core-pool-size=5

server-common/src/test/java/org/a2aproject/sdk/server/tasks/ResultAggregatorTest.java

@@ -62,7 +62,7 @@ public class ResultAggregatorTest {
     @BeforeEach
     void setUp() {
         MockitoAnnotations.openMocks(this);
-        aggregator = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor);
+        aggregator = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
     }
 
     // Helper methods for creating sample data
@@ -120,7 +120,7 @@ public void onTaskFinalized(String taskId) {
     @Test
     void testConstructorWithMessage() {
         Message initialMessage = createSampleMessage("initial", "msg1", Message.Role.ROLE_USER);
-        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, testExecutor, testExecutor);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
 
         // Test that the message is properly stored by checking getCurrentResult
         assertEquals(initialMessage, aggregatorWithMessage.getCurrentResult());
@@ -131,7 +131,7 @@ void testConstructorWithMessage() {
     @Test
     void testGetCurrentResultWithMessageSet() {
         Message sampleMessage = createSampleMessage("hola", "msg1", Message.Role.ROLE_USER);
-        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, testExecutor, testExecutor);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
 
         EventKind result = aggregatorWithMessage.getCurrentResult();
 
@@ -166,7 +166,7 @@ void testConstructorStoresTaskManagerCorrectly() {
 
     @Test
     void testConstructorWithNullMessage() {
-        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor);
+        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
         Task expectedTask = createSampleTask("null_msg_task", TaskState.TASK_STATE_WORKING, "ctx1");
         when(mockTaskManager.getTask()).thenReturn(expectedTask);
 
@@ -226,7 +226,7 @@ void testMultipleGetCurrentResultCalls() {
     void testGetCurrentResultWithMessageTakesPrecedence() {
         // Test that when both message and task are available, message takes precedence
         Message message = createSampleMessage("priority message", "pri1", Message.Role.ROLE_USER);
-        ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, testExecutor, testExecutor);
+        ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
 
         // Even if we set up the task manager to return something, message should take precedence
         Task task = createSampleTask("should_not_be_returned", TaskState.TASK_STATE_WORKING, "ctx1");
@@ -297,7 +297,7 @@ void testBlockingReturnsTaskWhenStoreBecomesVisibleAfterEmptyCapture() throws Ex
 
         TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null);
         ResultAggregator blockingAggregator =
-                new ResultAggregator(taskManager, null, testExecutor, testExecutor);
+                new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
 
         MainEventBus mainEventBus = new MainEventBus();
         InMemoryQueueManager queueManager =
@@ -325,7 +325,7 @@ void testBlockingStillRaisesMissingTaskErrorWhenStoreRemainsEmpty() {
         TaskStore taskStore = mock(TaskStore.class);
         TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null);
         ResultAggregator blockingAggregator =
-                new ResultAggregator(taskManager, null, testExecutor, testExecutor);
+                new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.MILLISECONDS.toNanos(100));
 
         InternalError error = assertThrows(InternalError.class, () ->
                 blockingAggregator.consumeAndBreakOnInterrupt(createClosedEventConsumer(taskId), true));
@@ -339,7 +339,7 @@ void testNonBlockingDoesNotPollTaskStoreWhenCaptureIsEmpty() {
         TaskStore taskStore = mock(TaskStore.class);
         TaskManager taskManager = new TaskManager(taskId, "ctx1", taskStore, null);
         ResultAggregator nonBlockingAggregator =
-                new ResultAggregator(taskManager, null, testExecutor, testExecutor);
+                new ResultAggregator(taskManager, null, testExecutor, testExecutor, TimeUnit.SECONDS.toNanos(1));
 
         assertThrows(InternalError.class, () ->
                 nonBlockingAggregator.consumeAndBreakOnInterrupt(createClosedEventConsumer(taskId), false));