fix: synchronize ClientTaskManager to prevent race condition in concu… (#742)

…rrent SSE event processing

ClientTaskManager was not thread-safe, causing intermittent test
failures when
SSE events were delivered concurrently by the HTTP client. The race
condition
occurred when multiple events (startWork, addArtifact, complete) were
emitted
rapidly in streaming mode.

Without synchronization, concurrent threads could interleave
read-modify-write
operations on currentTask, resulting in lost updates. Specifically, the
artifact
update could be overwritten by a status update, leaving the final
COMPLETED task
with no artifacts. The fast time to failure of the test 0.017s,
indicates that the
COMPLETED status update was received, and that the latch did not time
out.

This manifested as intermittent failures in agent-to-agent tests where
the test
received a COMPLETED task but extractTextFromTask() returned an empty
string
because the artifact list was empty.

Fix: Added synchronized keyword to all ClientTaskManager methods
(getCurrentTask,
saveTaskEvent variants, updateWithMessage) to ensure atomic processing
of events
regardless of HTTP client threading behavior.

Impact: All client transports (JSON-RPC, gRPC, REST) in streaming mode
Validated: 100 consecutive test iterations passed on CI

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: kabir

client/base/src/main/java/io/a2a/client/ClientTaskManager.java

@@ -35,22 +35,22 @@ public ClientTaskManager() {
         this.contextId = null;
     }
 
-    public Task getCurrentTask() throws A2AClientInvalidStateError {
+    public synchronized Task getCurrentTask() throws A2AClientInvalidStateError {
         if (currentTask == null) {
             throw new A2AClientInvalidStateError("No current task");
         }
         return currentTask;
     }
 
-    public Task saveTaskEvent(Task task) throws A2AClientInvalidArgsError {
+    public synchronized Task saveTaskEvent(Task task) throws A2AClientInvalidArgsError {
         if (currentTask != null) {
             throw new A2AClientInvalidArgsError("Task is already set, create new manager for new tasks.");
         }
         saveTask(task);
         return task;
     }
 
-    public Task saveTaskEvent(TaskStatusUpdateEvent taskStatusUpdateEvent) throws A2AClientError {
+    public synchronized Task saveTaskEvent(TaskStatusUpdateEvent taskStatusUpdateEvent) throws A2AClientError {
         if (taskId == null) {
             taskId = taskStatusUpdateEvent.taskId();
         }
@@ -86,7 +86,7 @@ public Task saveTaskEvent(TaskStatusUpdateEvent taskStatusUpdateEvent) throws A2
         return currentTask;
     }
 
-    public Task saveTaskEvent(TaskArtifactUpdateEvent taskArtifactUpdateEvent) {
+    public synchronized Task saveTaskEvent(TaskArtifactUpdateEvent taskArtifactUpdateEvent) {
         if (taskId == null) {
             taskId = taskArtifactUpdateEvent.taskId();
         }
@@ -113,7 +113,7 @@ public Task saveTaskEvent(TaskArtifactUpdateEvent taskArtifactUpdateEvent) {
      * @param task the task to update
      * @return the updated task
      */
-    public Task updateWithMessage(Message message, Task task) {
+    public synchronized Task updateWithMessage(Message message, Task task) {
         Task.Builder taskBuilder = Task.builder(task);
         List<Message> history = new ArrayList<>(task.history());
         if (task.status().message() != null) {