fix(server): support asynchronous agent execution via keepAlive

[malladinagarjuna2]

Overview

This PR introduces native support for completely asynchronous agent execution pipelines (e.g., RxJava Single, Kotlin Coroutines, or generic background thread-hopping) to solve premature event queue closures.

Fixes #872

The Problem

Previously, the DefaultRequestHandler assumed an agent was completely finished as soon as the AgentExecutor.execute() method returned. If the agent didn't emit a terminal event, the server enforced a strict 1.5-second timeout and forcefully closed the event queue to prevent leaks.
When users built ADK tools that utilized reactive thread-hopping (like an RxJava Single on an I/O dispatcher), the execute() block would return immediately, triggering the teardown before the background thread had a chance to emit its final event.

The Solution

1. AgentEmitter: Added a keepAlive() flag. Agents can invoke this to explicitly inform the server that the execution lifecycle has been handed off to an asynchronous thread.
2. EnhancedRunnable: Surfaced the AgentEmitter so it can be dynamically inspected by the orchestration loop.
3. DefaultRequestHandler: Re-evaluates the async state after the agent returns. If keepAlive() was invoked, the handler skips the early queue closure timeout, allowing the EventConsumer to wait naturally for the background thread to emit the terminal event.

Testing

• Added testAsyncAgentWithKeepAlive_Blocking_WaitsForCompletion to DefaultRequestHandlerTest to simulate a background thread executing and verifying that the client request blocks securely without timing out or closing the queue prematurely.
• Ran full mvn clean install verifying complete backward compatibility with existing synchronous agents.

[malladinagarjuna2]

@kabir could you please verify

[kabir]

Hi thanks for the PR :-)

There are quite many in the queue right now, so I'm getting some help from Claude in reviewing them in bulk. Apologies if something isn't clear, and feel free to discuss further.

What the problem is

The bug is real. When an agent's execute() returns immediately but hands off real work to a background thread (e.g., RxJava
Single.subscribeOn(Schedulers.io())), the server treats execute() returning as "agent is done." The whenComplete callback (line
1085) fires immediately and invokes doneCallbacks, which sets agentCompleted = true in the EventConsumer. The EventConsumer then
sees an empty queue + agentCompleted and closes the queue after 1.5 seconds (3 x 500ms poll timeouts at line 149-155). By the time
the background thread calls emitter.complete(), the ChildQueue is already gone.

Result: blocking calls throw InternalError, streaming calls return empty.
Result: blocking calls throw InternalError, streaming calls return empty.

What the PR does

Adds a keepAlive() method to AgentEmitter. When called by the agent inside execute(), it sets an isAsync flag. The whenComplete callback then
skips invokeDoneCallbacks(), so agentCompleted is never set, and the EventConsumer keeps polling until the background thread emits a terminal
event.

Concerns

1. keepAlive() is unnecessary — terminalStateReached already exists. AgentEmitter already has a terminalStateReached AtomicBoolean (line 104)
   that gets set when complete(), fail(), or cancel() is called. The whenComplete callback could simply check emitter.terminalStateReached instead
   of requiring agents to call a new API method. If execute() returned without reaching terminal state, the agent must be async — no opt-in needed.
   This would fix the problem transparently without any API surface change.

2. No timeout fallback for async agents. If the background thread crashes or hangs without calling emitter.complete(), the EventConsumer polls
   forever. There's no maximum timeout. The existing 1.5s grace period was a safety net; this PR removes it without a replacement.

3. invokeDoneCallbacks() suppression is too broad. By skipping ALL done callbacks, the PR suppresses not just the agentCompleted signal but any
   other cleanup registered via addDoneCallback. When async agents do eventually complete, done callbacks are never invoked at all — they just
   silently disappear.

4. Repeated TOCTOU-style checks. producerRunnable.getEmitter() != null && producerRunnable.getEmitter().isAsync() appears 3 times. Between the
   null check and the isAsync() call, the volatile field could theoretically change (and it's just verbose regardless).

5. The blocking path (fire-and-forget) may not work correctly. In the onMessageSend fire-and-forget path, the PR skips queue.close() for async
   agents. The consumption future (consumptionFuture.get()) still has its own timeout, but if the background thread takes longer than
   consumptionCompletionTimeoutSeconds, it throws InternalError — which may not be the right behavior.

6. Comments are excessive. The PR adds several multi-line comments explaining what keepAlive() does and why checks are re-evaluated. The code
   would be clearer with better naming (e.g., just checking terminalStateReached) than with explanatory comments.

---

In short: The bug is real and worth fixing, but the approach of adding a new keepAlive() API method is the wrong solution. The
terminalStateReached flag already on AgentEmitter provides the exact same signal without requiring agent authors to remember to call anything.
The fix should use that existing mechanism and also add a fallback timeout for safety.

[kabir]

See above comment