From 94f0d7f54f50ed631f001d530d0782ca93e9c45a Mon Sep 17 00:00:00 2001 From: chopmob-cloud Date: Mon, 13 Jul 2026 19:23:33 +0100 Subject: [PATCH] fix(server): close event stream when requiresInput/requiresAuth force isFinal AgentEmitter#requiresInput(message, isFinal=true) and requiresAuth(message, isFinal=true) set the internal terminalStateReached flag but never completed the stream: the emitted TaskStatusUpdateEvent#isFinal() is derived from the (non-terminal) task state, so EventConsumer kept the SSE / Flow.Publisher open. When the caller forces finality on a non-terminal (interrupted) state, enqueue a QueueClosedEvent so the stream completes after the status event is delivered, letting the client resume via a new request. Interrupted states still stay open by default (#756). Also removes a stale .isFinal(false) call from the emitEvent Javadoc; the builder no longer exposes it. Fixes #975 --- .../sdk/server/tasks/AgentEmitter.java | 11 ++++++++++- .../sdk/server/tasks/AgentEmitterTest.java | 19 +++++++++++++++---- 2 files changed, 25 insertions(+), 5 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java index b15e99116..5e97db848 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java @@ -9,6 +9,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.a2aproject.sdk.server.events.EventQueue; +import org.a2aproject.sdk.server.events.QueueClosedEvent; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.Artifact; import org.a2aproject.sdk.spec.Event; @@ -155,6 +156,15 @@ private void updateStatus(TaskState state, @Nullable Message message, boolean is .status(new TaskStatus(state, message, null)) .build(); eventQueue.enqueueEvent(event); + + // If the caller forced finality on a non-terminal (interrupted) state, the + // event's own isFinal() is derived from the state and is false, so the stream + // would otherwise stay open (interrupted states are intentionally non-terminal; + // see #756). Signal explicit closure so the stream completes after this event + // is delivered, letting the client resume via a new request. + if (isFinal && !state.isFinal()) { + eventQueue.enqueueEvent(new QueueClosedEvent(taskId)); + } } /** @@ -569,7 +579,6 @@ public void addTask(Task task) { * .taskId(context.getTaskId()) * .contextId(context.getContextId()) * .status(new TaskStatus(TaskState.WORKING)) - * .isFinal(false) * .build(); * emitter.emitEvent(event); * } diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java index 3799b3e1b..ecb98922e 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterTest.java @@ -14,6 +14,7 @@ import org.a2aproject.sdk.server.agentexecution.RequestContext; import org.a2aproject.sdk.server.events.EventQueue; +import org.a2aproject.sdk.server.events.QueueClosedEvent; import org.a2aproject.sdk.server.events.EventQueueItem; import org.a2aproject.sdk.server.events.EventQueueUtil; import org.a2aproject.sdk.server.events.InMemoryQueueManager; @@ -187,13 +188,13 @@ public void testRequiresInputWithMessage() throws Exception { @Test public void testRequiresInputWithFinalTrue() throws Exception { agentEmitter.requiresInput(true); - checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_INPUT_REQUIRED, null); + checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_INPUT_REQUIRED, null, true); } @Test public void testRequiresInputWithMessageAndFinalTrue() throws Exception { agentEmitter.requiresInput(SAMPLE_MESSAGE, true); - checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_INPUT_REQUIRED, SAMPLE_MESSAGE); + checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_INPUT_REQUIRED, SAMPLE_MESSAGE, true); } @Test @@ -211,13 +212,13 @@ public void testRequiresAuthWithMessage() throws Exception { @Test public void testRequiresAuthWithFinalTrue() throws Exception { agentEmitter.requiresAuth(true); - checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_AUTH_REQUIRED, null); + checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_AUTH_REQUIRED, null, true); } @Test public void testRequiresAuthWithMessageAndFinalTrue() throws Exception { agentEmitter.requiresAuth(SAMPLE_MESSAGE, true); - checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_AUTH_REQUIRED, SAMPLE_MESSAGE); + checkTaskStatusUpdateEventOnQueue(false, TaskState.TASK_STATE_AUTH_REQUIRED, SAMPLE_MESSAGE, true); } @Test @@ -486,6 +487,10 @@ public void sendMessageWithMismatchedContextIdThrows() { } private TaskStatusUpdateEvent checkTaskStatusUpdateEventOnQueue(boolean isFinal, TaskState state, Message statusMessage) throws Exception { + return checkTaskStatusUpdateEventOnQueue(isFinal, state, statusMessage, false); + } + + private TaskStatusUpdateEvent checkTaskStatusUpdateEventOnQueue(boolean isFinal, TaskState state, Message statusMessage, boolean expectStreamClose) throws Exception { // Wait up to 5 seconds for event (async MainEventBusProcessor needs time to distribute) EventQueueItem item = eventQueue.dequeueEventItem(WAIT_MILLI_SECONDS); assertNotNull(item); @@ -501,6 +506,12 @@ private TaskStatusUpdateEvent checkTaskStatusUpdateEventOnQueue(boolean isFinal, assertEquals(state, tsue.status().state()); assertEquals(statusMessage, tsue.status().message()); + if (expectStreamClose) { + EventQueueItem closeItem = eventQueue.dequeueEventItem(WAIT_MILLI_SECONDS); + assertNotNull(closeItem); + assertInstanceOf(QueueClosedEvent.class, closeItem.getEvent()); + } + // Check no additional events (still use 0 timeout for this check) assertNull(eventQueue.dequeueEventItem(0));