fix(server): close event stream when requiresInput/requiresAuth force isFinal

[chopmob-cloud]

Fixes #975.

Problem

AgentEmitter#requiresInput(message, isFinal=true) (and requiresAuth(..., isFinal=true)) set the internal terminalStateReached flag but never completed the stream. The emitted TaskStatusUpdateEvent#isFinal() is derived from the task state (status.state().isFinal()), which is false for the interrupted INPUT_REQUIRED / AUTH_REQUIRED states, so EventConsumer kept the SSE / Flow.Publisher open. The isFinal argument therefore only blocked further updates; it could not end the stream, and connections an agent asked to release stayed open.

Fix

When the caller forces finality on a non-terminal state, enqueue the existing QueueClosedEvent after the status event:

if (isFinal && !state.isFinal()) {
    eventQueue.enqueueEvent(new QueueClosedEvent(taskId));
}

EventConsumer already treats QueueClosedEvent as a stream-terminating event, so the status event is delivered and then the stream completes, letting the client resume via a new request. This uses the SDK's own close path and touches neither the event record nor the proto. Interrupted states still stay open by default (#756); only an explicit isFinal=true closes them. Terminal states are unaffected (their event's isFinal() is already true, and the !state.isFinal() guard skips them).

Also removes a stale .isFinal(false) call from the emitEvent Javadoc, since the builder no longer exposes it.

Tests

• Updated the four interrupted+final tests to assert the trailing QueueClosedEvent via a backward-compatible expectStreamClose overload of the existing helper.
• Verified on JDK 17: server-common suite 412/412 pass; with the fix reverted, the four forced-final tests fail (confirming they exercise the behavior).

Note on approach

This implements the "wire isFinal to actually close the stream" option. If you'd rather treat interrupted streams as always staying open and have isFinal on requiresInput/requiresAuth deprecated/documented instead, I'm happy to switch to that.

[gemini-code-assist]

Code Review

This pull request introduces explicit stream closure handling in AgentEmitter when finality is forced on a non-terminal state by enqueuing a QueueClosedEvent. This ensures the stream completes and allows clients to resume via a new request. The corresponding unit tests in AgentEmitterTest have been updated to verify that QueueClosedEvent is correctly emitted under these conditions. I have no feedback to provide as the implementation is correct and well-tested.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[kabir]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces the emission of a QueueClosedEvent in AgentEmitter when a final status is forced on a non-terminal (interrupted) task state, ensuring the event stream is explicitly closed so clients can resume via a new request. The corresponding unit tests in AgentEmitterTest have been updated to verify this behavior. I have no feedback to provide as there are no review comments.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[kabir]

Hi, sorry for the delay, there is a lot going on this week, and I'm on my own.

I looked at the code, since I didn't remember what 'forcing final' meant, and then I saw the AgentEmitter.requiresInput() and requiresAuth() methods have an overload with the extra isFinal parameter.

I don't remember adding that, or why I added it :-)

I asked Claude to check if this is something coming from the spec, and it says:

exactly these fields for TaskStatus:

• state (TaskState, required)
• message (Message)
• timestamp (Timestamp)

And TaskStatusUpdateEvent has:

• task_id, context_id, status, metadata

No boolean final anywhere.

The spec also clearly distinguishes:

• Terminal states: COMPLETED, FAILED, CANCELED, REJECTED
• Interrupted states: INPUT_REQUIRED, AUTH_REQUIRED

Interrupted states are explicitly not terminal — they pause execution awaiting further input/auth.

So the isFinal parameter on requiresInput() and requiresAuth() in AgentEmitter is a SDK invention with no basis in the
spec. It lets you mark an interrupted state as "final" (preventing further status updates), which contradicts the
spec's model where interrupted states are resumable.

Looks like it should be removed. Want me to clean that up?

So I wonder now if this was something which was accidentally added at some point.
Let's see what @ehsavoie and @jmesnil say when they come back.

[chopmob-cloud]

Thanks for taking a look, and no worries on timing.

I think the confusion is worth naming directly: isFinal here controls the stream/connection lifecycle, not the task state. Those are independent in A2A. INPUT_REQUIRED / AUTH_REQUIRED stay non-terminal and resumable in both cases, and TaskState#isFinal() is correctly false throughout, which this PR does not change. What isFinal=true decides is whether this SSE / Flow.Publisher stays open (client resumes on the same stream, the #756 default) or completes (client resumes via a new message/send or tasks/resubscribe). Closing the stream on an interrupted state is just the agent releasing the connection while it waits, not asserting a terminal task state. Before the fix it set the internal flag but never actually closed the stream, so the option was a no-op; this makes the close path real via the SDK's existing QueueClosedEvent.

That said, #975 from @bianjp listed both "make it close" and "remove/deprecate if unused" as acceptable, so the API-surface call is genuinely yours. If you would rather drop the isFinal overloads to keep the surface minimal, that also resolves #975 and I am happy to close this PR. If the "release the connection on an interrupted state" capability is worth keeping, this makes it behave as documented. Either way it resolves the reported bug, and I am glad to follow whichever direction you and @ehsavoie / @jmesnil settle on.