From e1185bbcc4095c9cae51049a47410126651aadf2 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Mon, 10 Aug 2026 20:25:55 +0800 Subject: [PATCH 1/6] fix: enforce task state-machine transitions and serialize cancels --- .../DefaultRequestHandler.java | 21 ++++ .../sdk/server/tasks/TaskManager.java | 44 +++++++ .../DefaultRequestHandlerTest.java | 74 +++++++++++ .../sdk/server/tasks/TaskManagerTest.java | 119 ++++++++++++++++++ 4 files changed, 258 insertions(+) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index d41b1e9c0..7197b6036 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -257,6 +257,18 @@ public class DefaultRequestHandler implements RequestHandler { private final ConcurrentMap> runningAgents = new ConcurrentHashMap<>(); + /** + * Per-task lock registry serializing {@link #onCancelTask} check-then-act sequences. + *

+ * The cancel path reads the task, checks it is not already in a terminal state, and + * then invokes the agent executor to enqueue the CANCELED event. Without a lock, two + * concurrent cancels (or a cancel racing a concurrent completion) could both observe + * the pre-transition state and both act on it (BUG-44). Entries are retained for the + * lifetime of the JVM; the registry is bounded by the number of distinct tasks that + * have ever been canceled, in the same way the in-memory task store is unbounded. + */ + private final ConcurrentMap cancelLocks = new ConcurrentHashMap<>(); + private Executor executor; private Executor eventConsumerExecutor; @@ -466,6 +478,15 @@ public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext con @Override public Task onCancelTask(CancelTaskParams params, ServerCallContext context) throws A2AError { + // Serialize check-then-act per task so two concurrent cancels (or a cancel racing + // a concurrent completion) cannot both act on the pre-transition state (BUG-44). + Object cancelLock = cancelLocks.computeIfAbsent(params.id(), k -> new Object()); + synchronized (cancelLock) { + return doCancelTask(params, context); + } + } + + private Task doCancelTask(CancelTaskParams params, ServerCallContext context) throws A2AError { Task task = taskStore.get(params.id()); if (task == null) { throw new TaskNotFoundError(); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java index e7a8b1568..e7f7bdd1b 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java @@ -19,6 +19,7 @@ import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskState; import org.a2aproject.sdk.spec.TaskStatus; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; import org.jspecify.annotations.Nullable; @@ -69,6 +70,13 @@ boolean saveTaskEvent(Task task, boolean isReplicated) throws A2AServerException boolean saveTaskEvent(Task task, boolean isReplicated, @Nullable AtomicReference taskSnapshot) throws A2AServerException { checkIdsAndUpdateIfNecessary(task.id(), task.contextId()); + // Defensive state-machine check: a task that already reached a terminal state must + // not be overwritten by a task snapshot carrying a different state (BUG-43). + Task current = getTask(); + if (current != null && current.status() != null && current.status().state() != null + && task.status() != null && task.status().state() != null) { + validateStateTransition(current.status().state(), task.status().state(), task.id()); + } Task savedTask = saveTask(task, isReplicated); if (taskSnapshot != null) { taskSnapshot.set(savedTask); @@ -85,6 +93,12 @@ boolean saveTaskEvent(TaskStatusUpdateEvent event, boolean isReplicated, @Nullab checkIdsAndUpdateIfNecessary(event.taskId(), event.contextId()); Task task = ensureTask(event.taskId(), event.contextId()); + // State-machine validation: reject transitions that would overwrite a terminal + // state with a different state (BUG-43). Re-arriving events carrying the same + // final state remain allowed (idempotent replays / replication). + TaskState currentState = task.status() != null ? task.status().state() : null; + TaskState newState = event.status() != null ? event.status().state() : null; + validateStateTransition(currentState, newState, event.taskId()); Task.Builder builder = Task.builder(task) .status(event.status()); @@ -232,6 +246,36 @@ private Task ensureTask(String eventTaskId, String eventContextId) { return task; } + /** + * Validates a task state transition before it is persisted (BUG-43). + *

+ * A terminal (final) state must not be overwritten by a different state: + * once a task is {@code COMPLETED}/{@code FAILED}/{@code CANCELED}/{@code REJECTED} + * it stays in that state. Events re-arriving with the same final state are + * allowed, so replicated replays and idempotent retries keep working. + *

+ * Transitions from any non-terminal state to any state are permitted (e.g. + * SUBMITTED → WORKING → COMPLETED/FAILED/CANCELED, interrupted-state resume flows), + * matching the transitions the A2A spec and the reference agents exercise. + * + * @param currentState the task's current state, or {@code null} if unknown + * @param newState the state requested by the event, or {@code null} if unknown + * @param taskId the task identifier, used in the error message + * @throws A2AServerException if the transition would overwrite a terminal state + */ + private static void validateStateTransition(@Nullable TaskState currentState, @Nullable TaskState newState, + String taskId) throws A2AServerException { + if (currentState == null || newState == null) { + return; + } + if (currentState.isFinal() && currentState != newState) { + throw new A2AServerException( + "Task " + taskId + " is already in terminal state " + currentState + + " and cannot transition to " + newState, + new InternalError("Task " + taskId + " is already in terminal state " + currentState)); + } + } + private Task createTask(String taskId, String contextId) { List history = initialMessage != null ? List.of(initialMessage) : Collections.emptyList(); return Task.builder() diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index cf768231e..95909a4c8 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -13,10 +13,14 @@ import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Flow; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; import org.a2aproject.sdk.server.ServerCallContext; @@ -36,6 +40,7 @@ import org.a2aproject.sdk.server.tasks.PushNotificationSender; import org.a2aproject.sdk.server.tasks.TaskStore; import org.a2aproject.sdk.spec.A2AError; +import org.a2aproject.sdk.spec.CancelTaskParams; import org.a2aproject.sdk.spec.Event; import org.a2aproject.sdk.spec.EventKind; import org.a2aproject.sdk.spec.InvalidParamsError; @@ -46,6 +51,7 @@ import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; import org.a2aproject.sdk.spec.TaskNotFoundError; +import org.a2aproject.sdk.spec.TaskNotCancelableError; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; import org.a2aproject.sdk.spec.TaskState; import org.a2aproject.sdk.spec.TaskStatus; @@ -1149,6 +1155,7 @@ public void onComplete() { } @Test + void testOnGetTaskHistoryLengthLimitsHistory() throws Exception { Task task = taskWithHistory("task-hl-limit"); taskStore.save(task, false); @@ -1185,5 +1192,72 @@ private Task taskWithHistory(String id) { .parts(new TextPart("three")).build())) .artifacts(List.of()) .build(); + +void testConcurrentCancelsAreSerialized() throws Exception { + // BUG-44 regression: two concurrent cancels of the same task must serialize on a + // per-task lock so the second one observes the CANCELED terminal state and fails + // with TaskNotCancelableError instead of both acting on the pre-transition state. + Task workingTask = Task.builder() + .id("task-cancel-lock") + .contextId("ctx-cancel") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .history(List.of()) + .artifacts(List.of()) + .build(); + taskStore.save(workingTask, false); + + CountDownLatch cancelEntered = new CountDownLatch(1); + CountDownLatch releaseCancel = new CountDownLatch(1); + + agentExecutorCancel = (context, emitter) -> { + cancelEntered.countDown(); + try { + releaseCancel.await(10, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + emitter.cancel(); + }; + + ExecutorService cancelExec = Executors.newFixedThreadPool(2); + try { + CountDownLatch firstDone = new CountDownLatch(1); + Future first = cancelExec.submit(() -> { + try { + Task result = requestHandler.onCancelTask( + new CancelTaskParams("task-cancel-lock"), NULL_CONTEXT); + firstDone.countDown(); + return result; + } catch (A2AError e) { + firstDone.countDown(); + throw e; + } + }); + + // Wait until the first cancel is inside agentExecutor.cancel() (holding the per-task lock) + assertTrue(cancelEntered.await(5, TimeUnit.SECONDS), + "First cancel should enter agentExecutor.cancel()"); + + // The second cancel must block on the per-task lock while the first is in progress + Future second = cancelExec.submit(() -> requestHandler.onCancelTask( + new CancelTaskParams("task-cancel-lock"), NULL_CONTEXT)); + assertThrows(TimeoutException.class, () -> second.get(300, TimeUnit.MILLISECONDS), + "Second cancel should not complete while the first holds the per-task lock"); + + // Release the first cancel so it can enqueue CANCELED and finish + releaseCancel.countDown(); + assertTrue(firstDone.await(10, TimeUnit.SECONDS), "First cancel should complete"); + assertEquals(TaskState.TASK_STATE_CANCELED, first.get().status().state()); + + // The second cancel now observes the terminal state and is rejected + ExecutionException ex = assertThrows(ExecutionException.class, second::get); + assertInstanceOf(TaskNotCancelableError.class, ex.getCause(), + "Second cancel should fail with TaskNotCancelableError"); + } finally { + releaseCancel.countDown(); + cancelExec.shutdownNow(); + } +>>>>>>> 9fcf9f46 +(fix: enforce task state-machine transitions and serialize cancels) } } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java index 75e04f2cd..13ab03074 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java @@ -739,4 +739,123 @@ public void testUpdateWithMessage() throws A2AServerException { assertEquals("task message", ((TextPart) updated.history().get(1).parts().get(0)).text()); assertEquals("update message", ((TextPart) updated.history().get(2).parts().get(0)).text()); } + + @Test + public void testRejectStatusUpdateOverwritingTerminalState() throws A2AServerException { + // Seed a COMPLETED task + Task completedTask = Task.builder() + .id("task-terminal") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + taskStore.save(completedTask, false); + TaskManager tm = new TaskManager("task-terminal", "ctx-1", taskStore, null); + + // A status update to a different state after the terminal state must be rejected (BUG-43) + TaskStatusUpdateEvent workingEvent = TaskStatusUpdateEvent.builder() + .taskId("task-terminal") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + assertThrows(A2AServerException.class, () -> tm.saveTaskEvent(workingEvent, false)); + + // The persisted task must remain in its terminal state + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-terminal").status().state()); + } + + @Test + public void testRejectStatusUpdateToDifferentTerminalState() throws A2AServerException { + Task completedTask = Task.builder() + .id("task-terminal-2") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + taskStore.save(completedTask, false); + TaskManager tm = new TaskManager("task-terminal-2", "ctx-1", taskStore, null); + + // COMPLETED must not be overwritten by FAILED either + TaskStatusUpdateEvent failedEvent = TaskStatusUpdateEvent.builder() + .taskId("task-terminal-2") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_FAILED)) + .build(); + assertThrows(A2AServerException.class, () -> tm.saveTaskEvent(failedEvent, false)); + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-terminal-2").status().state()); + } + + @Test + public void testSameTerminalStateReplayAllowed() throws A2AServerException { + Task completedTask = Task.builder() + .id("task-replay") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + taskStore.save(completedTask, false); + TaskManager tm = new TaskManager("task-replay", "ctx-1", taskStore, null); + + // Idempotent replay of the same final state must remain allowed (replication/replay) + TaskStatusUpdateEvent completedAgain = TaskStatusUpdateEvent.builder() + .taskId("task-replay") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + tm.saveTaskEvent(completedAgain, false); + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-replay").status().state()); + } + + @Test + public void testRejectTaskEventOverwritingTerminalState() throws A2AServerException { + Task completedTask = Task.builder() + .id("task-terminal-3") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + taskStore.save(completedTask, false); + TaskManager tm = new TaskManager("task-terminal-3", "ctx-1", taskStore, null); + + // A full Task snapshot carrying a different (non-terminal) state must be rejected + Task submittedSnapshot = Task.builder() + .id("task-terminal-3") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)) + .build(); + assertThrows(A2AServerException.class, () -> tm.saveTaskEvent(submittedSnapshot, false)); + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-terminal-3").status().state()); + } + + @Test + public void testNormalStateFlowAllowed() throws A2AServerException { + // SUBMITTED -> WORKING -> COMPLETED must keep working (BUG-43 must not break normal flows) + TaskManager tm = new TaskManager("task-flow", "ctx-1", taskStore, null); + + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-flow").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_SUBMITTED)).build(), false); + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-flow").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)).build(), false); + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-flow").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)).build(), false); + + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-flow").status().state()); + } + + @Test + public void testInterruptedStateResumeFlowAllowed() throws A2AServerException { + // INPUT_REQUIRED -> WORKING -> COMPLETED (resume flow) must keep working + TaskManager tm = new TaskManager("task-interrupted", "ctx-1", taskStore, null); + + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-interrupted").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_INPUT_REQUIRED)).build(), false); + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-interrupted").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)).build(), false); + tm.saveTaskEvent(TaskStatusUpdateEvent.builder() + .taskId("task-interrupted").contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)).build(), false); + + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-interrupted").status().state()); + } } From 57b606a289cec2d6f6d985ebff705e4659827c1f Mon Sep 17 00:00:00 2001 From: meraklbz Date: Mon, 10 Aug 2026 23:36:48 +0800 Subject: [PATCH 2/6] test: make parallel replication count test deterministic testParallelReplicationBehavior sent TASK_STATE_COMPLETED events from the replicated threads. A COMPLETED event processed mid-stream finalizes the task and closes the queue, so overlapping normal enqueues no longer trigger replication and the final count assertion became timing- dependent (observed 0/1/2/3/21 instead of 25, locally and in CI). Use a non-terminal state for the replicated events; the replication hook skips them via isReplicated() regardless of state, so the test's intent (normal enqueues replicate, replicated events do not) is unchanged while the outcome is now deterministic. --- .../replicated/core/ReplicatedQueueManagerTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java b/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java index e524a717d..daced98d2 100644 --- a/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java +++ b/extras/queue-manager-replicated/core/src/test/java/org/a2aproject/sdk/extras/queuemanager/replicated/core/ReplicatedQueueManagerTest.java @@ -370,7 +370,12 @@ public void onTaskFinalized(String tid) { TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() .taskId(taskId) // Use same taskId as queue .contextId("test-context") - .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + // Use a non-terminal state: a COMPLETED event processed mid-stream + // finalizes the task and closes the queue, so overlapping normal + // enqueues no longer trigger replication and the count assertion + // becomes timing-dependent (flaky). Replicated events are skipped by + // the replication hook via isReplicated() regardless of state. + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) .build(); ReplicatedEventQueueItem replicatedEvent = new ReplicatedEventQueueItem(taskId, event); queueManager.onReplicatedEvent(replicatedEvent); From c2f70278f7c0da00a6de2ab0574aedd0e7e01346 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Tue, 11 Aug 2026 21:34:29 +0800 Subject: [PATCH 3/6] chore: remove internal tracking ids from comments --- .../sdk/server/requesthandlers/DefaultRequestHandler.java | 4 ++-- .../java/org/a2aproject/sdk/server/tasks/TaskManager.java | 6 +++--- .../server/requesthandlers/DefaultRequestHandlerTest.java | 8 +++++--- .../org/a2aproject/sdk/server/tasks/TaskManagerTest.java | 4 ++-- 4 files changed, 12 insertions(+), 10 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 7197b6036..4bf9e7a3f 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -263,7 +263,7 @@ public class DefaultRequestHandler implements RequestHandler { * The cancel path reads the task, checks it is not already in a terminal state, and * then invokes the agent executor to enqueue the CANCELED event. Without a lock, two * concurrent cancels (or a cancel racing a concurrent completion) could both observe - * the pre-transition state and both act on it (BUG-44). Entries are retained for the + * the pre-transition state and both act on it. Entries are retained for the * lifetime of the JVM; the registry is bounded by the number of distinct tasks that * have ever been canceled, in the same way the in-memory task store is unbounded. */ @@ -479,7 +479,7 @@ public ListTasksResult onListTasks(ListTasksParams params, ServerCallContext con @Override public Task onCancelTask(CancelTaskParams params, ServerCallContext context) throws A2AError { // Serialize check-then-act per task so two concurrent cancels (or a cancel racing - // a concurrent completion) cannot both act on the pre-transition state (BUG-44). + // a concurrent completion) cannot both act on the pre-transition state. Object cancelLock = cancelLocks.computeIfAbsent(params.id(), k -> new Object()); synchronized (cancelLock) { return doCancelTask(params, context); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java index e7f7bdd1b..051ded06a 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java @@ -71,7 +71,7 @@ boolean saveTaskEvent(Task task, boolean isReplicated, @Nullable AtomicReference throws A2AServerException { checkIdsAndUpdateIfNecessary(task.id(), task.contextId()); // Defensive state-machine check: a task that already reached a terminal state must - // not be overwritten by a task snapshot carrying a different state (BUG-43). + // not be overwritten by a task snapshot carrying a different state. Task current = getTask(); if (current != null && current.status() != null && current.status().state() != null && task.status() != null && task.status().state() != null) { @@ -94,7 +94,7 @@ boolean saveTaskEvent(TaskStatusUpdateEvent event, boolean isReplicated, @Nullab Task task = ensureTask(event.taskId(), event.contextId()); // State-machine validation: reject transitions that would overwrite a terminal - // state with a different state (BUG-43). Re-arriving events carrying the same + // state with a different state. Re-arriving events carrying the same // final state remain allowed (idempotent replays / replication). TaskState currentState = task.status() != null ? task.status().state() : null; TaskState newState = event.status() != null ? event.status().state() : null; @@ -247,7 +247,7 @@ private Task ensureTask(String eventTaskId, String eventContextId) { } /** - * Validates a task state transition before it is persisted (BUG-43). + * Validates a task state transition before it is persisted. *

* A terminal (final) state must not be overwritten by a different state: * once a task is {@code COMPLETED}/{@code FAILED}/{@code CANCELED}/{@code REJECTED} diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 95909a4c8..d61794ea1 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -1155,7 +1155,6 @@ public void onComplete() { } @Test - void testOnGetTaskHistoryLengthLimitsHistory() throws Exception { Task task = taskWithHistory("task-hl-limit"); taskStore.save(task, false); @@ -1195,6 +1194,11 @@ private Task taskWithHistory(String id) { void testConcurrentCancelsAreSerialized() throws Exception { // BUG-44 regression: two concurrent cancels of the same task must serialize on a + } + + void testConcurrentCancelsAreSerialized() throws Exception { + // Regression: two concurrent cancels of the same task must serialize on a + (chore: remove internal tracking ids from comments) // per-task lock so the second one observes the CANCELED terminal state and fails // with TaskNotCancelableError instead of both acting on the pre-transition state. Task workingTask = Task.builder() @@ -1257,7 +1261,5 @@ void testConcurrentCancelsAreSerialized() throws Exception { releaseCancel.countDown(); cancelExec.shutdownNow(); } ->>>>>>> 9fcf9f46 -(fix: enforce task state-machine transitions and serialize cancels) } } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java index 13ab03074..8aacf44a1 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java @@ -751,7 +751,7 @@ public void testRejectStatusUpdateOverwritingTerminalState() throws A2AServerExc taskStore.save(completedTask, false); TaskManager tm = new TaskManager("task-terminal", "ctx-1", taskStore, null); - // A status update to a different state after the terminal state must be rejected (BUG-43) + // A status update to a different state after the terminal state must be rejected TaskStatusUpdateEvent workingEvent = TaskStatusUpdateEvent.builder() .taskId("task-terminal") .contextId("ctx-1") @@ -825,7 +825,7 @@ public void testRejectTaskEventOverwritingTerminalState() throws A2AServerExcept @Test public void testNormalStateFlowAllowed() throws A2AServerException { - // SUBMITTED -> WORKING -> COMPLETED must keep working (BUG-43 must not break normal flows) + // SUBMITTED -> WORKING -> COMPLETED must keep working TaskManager tm = new TaskManager("task-flow", "ctx-1", taskStore, null); tm.saveTaskEvent(TaskStatusUpdateEvent.builder() From fffb8444641b4888093b540e97700f98316bea98 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Wed, 12 Aug 2026 07:57:10 +0800 Subject: [PATCH 4/6] fix: skip FAILED transition when an A2AError arrives for a terminal task Per the review: TaskManager.process() synthesized a TASK_STATE_FAILED event for every A2AError, which the state-machine guard then rejected for already-terminal tasks (terminal -> FAILED). The exception propagated through MainEventBusProcessor and clients received a misleading 'TaskStore persistence failed' InternalError instead of the original A2AError. If the task is already terminal, skip the state update entirely and let the A2AError signal finality to clients. Added regression tests for both the terminal and non-terminal A2AError paths. --- .../sdk/server/tasks/TaskManager.java | 29 +++++++++++---- .../sdk/server/tasks/TaskManagerTest.java | 37 +++++++++++++++++++ 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java index 051ded06a..3a206f923 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java @@ -182,13 +182,28 @@ public boolean process(Event event, boolean isReplicated, @Nullable AtomicRefere // Only create status update if we have contextId if (errorContextId != null) { - LOGGER.debug("A2AError event detected, transitioning task {} to FAILED", taskId); - TaskStatusUpdateEvent failedEvent = TaskStatusUpdateEvent.builder() - .taskId(taskId) - .contextId(errorContextId) - .status(new TaskStatus(TASK_STATE_FAILED)) - .build(); - isFinal = saveTaskEvent(failedEvent, isReplicated, taskSnapshot); + // If the task is already in a terminal state, skip the state + // update entirely: the synthesized FAILED event would be rejected + // by the state-machine guard (terminal -> FAILED is a terminal-to- + // different-terminal transition), and the resulting exception would + // surface to clients as a misleading internal error. The A2AError + // itself still signals finality to clients. + Task existingTask = getTask(); + TaskState currentState = existingTask != null && existingTask.status() != null + ? existingTask.status().state() : null; + if (currentState != null && currentState.isFinal()) { + LOGGER.debug("A2AError event for task {} already in terminal state {} - skipping state update", + taskId, currentState); + isFinal = true; + } else { + LOGGER.debug("A2AError event detected, transitioning task {} to FAILED", taskId); + TaskStatusUpdateEvent failedEvent = TaskStatusUpdateEvent.builder() + .taskId(taskId) + .contextId(errorContextId) + .status(new TaskStatus(TASK_STATE_FAILED)) + .build(); + isFinal = saveTaskEvent(failedEvent, isReplicated, taskSnapshot); + } } else { // Can't update status without contextId, but error is still terminal LOGGER.debug("A2AError event for task {} without contextId - skipping state update", taskId); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java index 8aacf44a1..e36ad6cc9 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/TaskManagerTest.java @@ -15,6 +15,7 @@ import java.util.Map; import org.a2aproject.sdk.spec.A2AServerException; +import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.Artifact; import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.Task; @@ -858,4 +859,40 @@ public void testInterruptedStateResumeFlowAllowed() throws A2AServerException { assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-interrupted").status().state()); } + + @Test + public void testA2AErrorOnTerminalTaskSkipsStateUpdate() throws A2AServerException { + // Seed a COMPLETED task + Task completedTask = Task.builder() + .id("task-a2aerror-terminal") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED)) + .build(); + taskStore.save(completedTask, false); + TaskManager tm = new TaskManager("task-a2aerror-terminal", "ctx-1", taskStore, null); + + // An A2AError for an already-terminal task must not attempt the FAILED + // transition (which the state machine rejects); process() returns true + // (final) without throwing and the terminal state is preserved. + A2AError error = new A2AError(-32603, "agent failed", null); + assertTrue(tm.process(error, false)); + + assertEquals(TaskState.TASK_STATE_COMPLETED, taskStore.get("task-a2aerror-terminal").status().state()); + } + + @Test + public void testA2AErrorOnNonTerminalTaskTransitionsToFailed() throws A2AServerException { + Task workingTask = Task.builder() + .id("task-a2aerror-working") + .contextId("ctx-1") + .status(new TaskStatus(TaskState.TASK_STATE_WORKING)) + .build(); + taskStore.save(workingTask, false); + TaskManager tm = new TaskManager("task-a2aerror-working", "ctx-1", taskStore, null); + + A2AError error = new A2AError(-32603, "agent failed", null); + assertTrue(tm.process(error, false)); + + assertEquals(TaskState.TASK_STATE_FAILED, taskStore.get("task-a2aerror-working").status().state()); + } } From c8e39a6411486a9695d85a95cf7ec49d585a6c19 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Wed, 12 Aug 2026 08:13:36 +0800 Subject: [PATCH 5/6] test: fix conflict-resolution assembly in DefaultRequestHandlerTest The rebase onto the updated upstream merged the history-length tests with the concurrent-cancel regression test; repair the assembled method signature and annotation after the merge. --- .../server/requesthandlers/DefaultRequestHandlerTest.java | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index d61794ea1..48205fe7b 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -1191,14 +1191,11 @@ private Task taskWithHistory(String id) { .parts(new TextPart("three")).build())) .artifacts(List.of()) .build(); - -void testConcurrentCancelsAreSerialized() throws Exception { - // BUG-44 regression: two concurrent cancels of the same task must serialize on a } + @Test void testConcurrentCancelsAreSerialized() throws Exception { // Regression: two concurrent cancels of the same task must serialize on a - (chore: remove internal tracking ids from comments) // per-task lock so the second one observes the CANCELED terminal state and fails // with TaskNotCancelableError instead of both acting on the pre-transition state. Task workingTask = Task.builder() From 000ad38965a40ca360b5f6b4b6b0b8a76b4d95a9 Mon Sep 17 00:00:00 2001 From: meraklbz Date: Wed, 12 Aug 2026 21:49:39 +0800 Subject: [PATCH 6/6] refactor: consolidate terminal-state handling per review - TaskManager.process(): instead of duplicating the terminal-state check, attempt the synthesized FAILED transition and catch the A2AServerException from validateStateTransition (the single authority). An A2AError arriving for a terminal task is treated as final without a state update. - onCancelTask: drop the per-task cancelLocks entry after doCancelTask returns (the task is then terminal; the state-machine guard rejects any future cancel without the lock), preventing unbounded map growth. --- .../DefaultRequestHandler.java | 11 +++++- .../sdk/server/tasks/TaskManager.java | 34 ++++++++----------- 2 files changed, 24 insertions(+), 21 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 4bf9e7a3f..4b63c6716 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -482,7 +482,16 @@ public Task onCancelTask(CancelTaskParams params, ServerCallContext context) thr // a concurrent completion) cannot both act on the pre-transition state. Object cancelLock = cancelLocks.computeIfAbsent(params.id(), k -> new Object()); synchronized (cancelLock) { - return doCancelTask(params, context); + try { + return doCancelTask(params, context); + } finally { + // After doCancelTask returns (successfully or via TaskNotCancelableError), + // the task is in a terminal state, so no future cancel can succeed — the + // state-machine guard rejects it even without the lock. Drop the entry to + // avoid unbounded growth of the map. The 2-arg remove keeps a concurrently + // created newer lock entry (for a future, already-terminal task) intact. + cancelLocks.remove(params.id(), cancelLock); + } } } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java index 3a206f923..775975727 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/TaskManager.java @@ -182,27 +182,21 @@ public boolean process(Event event, boolean isReplicated, @Nullable AtomicRefere // Only create status update if we have contextId if (errorContextId != null) { - // If the task is already in a terminal state, skip the state - // update entirely: the synthesized FAILED event would be rejected - // by the state-machine guard (terminal -> FAILED is a terminal-to- - // different-terminal transition), and the resulting exception would - // surface to clients as a misleading internal error. The A2AError - // itself still signals finality to clients. - Task existingTask = getTask(); - TaskState currentState = existingTask != null && existingTask.status() != null - ? existingTask.status().state() : null; - if (currentState != null && currentState.isFinal()) { - LOGGER.debug("A2AError event for task {} already in terminal state {} - skipping state update", - taskId, currentState); - isFinal = true; - } else { - LOGGER.debug("A2AError event detected, transitioning task {} to FAILED", taskId); - TaskStatusUpdateEvent failedEvent = TaskStatusUpdateEvent.builder() - .taskId(taskId) - .contextId(errorContextId) - .status(new TaskStatus(TASK_STATE_FAILED)) - .build(); + LOGGER.debug("A2AError event detected, transitioning task {} to FAILED", taskId); + TaskStatusUpdateEvent failedEvent = TaskStatusUpdateEvent.builder() + .taskId(taskId) + .contextId(errorContextId) + .status(new TaskStatus(TASK_STATE_FAILED)) + .build(); + try { isFinal = saveTaskEvent(failedEvent, isReplicated, taskSnapshot); + } catch (A2AServerException e) { + // Task already in a terminal state: the state-machine guard in + // validateStateTransition rejected the synthesized FAILED event + // (terminal -> FAILED). No state update is needed — the A2AError + // itself still signals finality to clients. + LOGGER.debug("A2AError for task {} already in terminal state - skipping state update", taskId); + isFinal = true; } } else { // Can't update status without contextId, but error is still terminal