Provide an API/hook to close streams after INPUT_REQUIRED / AUTH_REQUIRED

[bianjp]

Summary

Follow-up to #975 / #989.

After #756, interrupted states keep the stream open by design, and #989 correctly deprecated the misleading isFinal overloads on requiresInput / requiresAuth. We still need a supported way for server embeddings to complete the stream after emitting INPUT_REQUIRED / AUTH_REQUIRED.

Why

With SSE, only the server can complete the stream. The client can’t cleanly close it (dropping the connection still leaves server-side resources open until timeout). SSE is also one-way, so after INPUT_REQUIRED the client must send follow-up input via a new request anyway — keeping the original stream open doesn’t help that path, and unclosed connections become costly under load.

Prior discussion

• fix(server): close event stream when requiresInput/requiresAuth force isFinal #980 proposed wiring isFinal=true to close the stream; maintainers preferred not to, since isFinal isn’t in the spec and blurs task state vs connection lifecycle (comment).
• Suggested direction in [Bug]: AgentEmitter#requiresInput(..., isFinal=true) does not close the event stream #975: a hook on interrupted-state events plus a dedicated close API, so users can choose whether to keep the stream open or complete it.

Ask

Please consider adding an explicit close/hook API (exact shape TBD). Default can remain “stream stays open” per #756.

[kabir]

Hi, I definitely see the point in being able to have the user control how long SSE streams are kept open. But I disagree with the statement here:

With SSE, only the server can complete the stream. The client can’t cleanly close it (dropping the connection still leaves server-side resources open until timeout). SSE is also one-way, so after INPUT_REQUIRED the client must send follow-up input via a new request anyway — keeping the original stream open doesn’t help that path, and unclosed connections become costly under load.

I actually thought I'd posted on #975 yesterday, but it seems I forgot to hit the 'Comment' button 😀

Anyway, I pointed Claude at it who says:

When an SSE client disconnects, the ChildQueue IS cleaned up:

1. Vert.x fires HttpServerResponse.closeHandler in SseResponseWriter.java (line 84-88), which:

• Calls EventConsumer.cancel() — sets cancelled = true, polling loop exits within ~500ms
• Calls subscription.cancel() on the Reactive Streams subscription, which in
  DefaultRequestHandler.onMessageSendStream() (line 747) calls queue.close(true) — immediately clears the ChildQueue's
  LinkedBlockingDeque and drops all buffered events

2. MainQueue.childClosing() removes the ChildQueue from the children list, making it GC-eligible.

MainQueue stays open by design — with minimal overhead:

When all children disconnect and the task is NOT finalized, MainQueue.childClosing() (line 566-590) explicitly keeps
the MainQueue open. The resources held are:

• An empty CopyOnWriteArrayList (no ChildQueues)
• No threads — MainQueue has no thread of its own. The shared MainEventBusProcessor daemon thread calls
  distributeToChildren() which iterates an empty list and returns immediately
• No blocking queue — only ChildQueues have LinkedBlockingDeque; MainQueue submits directly to the shared MainEventBus
• Just the object itself in the ConcurrentHashMap of InMemoryQueueManager — a single Map entry

The issue's claim is inaccurate. The statement "dropping the connection still leaves server-side resources open until
timeout" is not how this SDK works. The SDK actively detects client disconnects via Vert.x's close handler and
promptly cleans up the ChildQueue and its resources. The MainQueue intentionally stays in the map to enable
resubscription (a new subscribeToTask can tap it), but that's a lightweight Map entry with no threads, no queues, and
no meaningful resource cost.

So the issue's "Why" section overstates the problem. The legitimate ask — providing an API/hook to let server
embedders complete the stream after interrupted states — still has value, but the resource-pressure argument doesn't
hold for this implementation.

But in any case I do see the need of being able to limit how long these stay alive, ie a Task might take days to complete

[bianjp]

Thanks for clarifying! My statement was largely influenced by my experience with Spring MVC's SseEmitter under the blocking Servlet (Tomcat) stack, where client disconnections aren't detected until the next write attempt or a timeout occurs (without explicit heartbeats).

I now realize this is a specific limitation of blocking I/O frameworks rather than a fundamental flaw in the SSE protocol itself or reactive implementations like WebFlux/Netty. Appreciate the insight!