fix(server): support asynchronous agent execution via keepAlive

server-common/src/main/java/org/a2aproject/sdk/server/agentexecution/AgentExecutor.java

@@ -97,6 +97,7 @@
  * @see org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler
  * @see org.a2aproject.sdk.spec.AgentCard
  */
+@FunctionalInterface
 public interface AgentExecutor {
     /**
      * Executes the agent's business logic for a message.
@@ -147,5 +148,7 @@ public interface AgentExecutor {
      * @throws org.a2aproject.sdk.spec.TaskNotCancelableError if this agent does not support cancellation
      * @throws A2AError if cancellation is supported but failed to execute
      */
-    void cancel(RequestContext context, AgentEmitter emitter) throws A2AError;
+    default void cancel(RequestContext context, AgentEmitter emitter) throws A2AError {
+        throw new org.a2aproject.sdk.spec.TaskNotCancelableError("Agent execution is not cancelable.");
+    }
 }

server-common/src/main/java/org/a2aproject/sdk/server/events/EnhancedRunnable.java

@@ -5,9 +5,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.jspecify.annotations.Nullable;
+import org.a2aproject.sdk.server.tasks.AgentEmitter;
 
 public abstract class EnhancedRunnable implements Runnable {
     private volatile @Nullable Throwable error;
+    private volatile @Nullable AgentEmitter emitter;
     private final List<DoneCallback> doneCallbacks = new CopyOnWriteArrayList<>();
     private final AtomicBoolean started = new AtomicBoolean(false);
 
@@ -19,6 +21,14 @@ public void setError(Throwable error) {
         this.error = error;
     }
 
+    public @Nullable AgentEmitter getEmitter() {
+        return emitter;
+    }
+
+    public void setEmitter(AgentEmitter emitter) {
+        this.emitter = emitter;
+    }
+
     public void addDoneCallback(DoneCallback doneCallback) {
         if (started.get()) {
             throw new IllegalStateException(

server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java

@@ -6,10 +6,12 @@
 import org.a2aproject.sdk.spec.A2AError;
 import org.a2aproject.sdk.spec.A2AServerException;
 import org.a2aproject.sdk.spec.Event;
+import org.a2aproject.sdk.spec.InternalError;
 import org.a2aproject.sdk.spec.Message;
 import org.a2aproject.sdk.spec.Task;
 import org.a2aproject.sdk.spec.TaskState;
 import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
+import org.a2aproject.sdk.server.tasks.AgentEmitter;
 import mutiny.zero.BackpressureStrategy;
 import mutiny.zero.TubeConfiguration;
 import mutiny.zero.ZeroPublisher;
@@ -27,6 +29,9 @@ public class EventConsumer {
     private volatile int pollTimeoutsAfterAgentCompleted = 0;
     private volatile @Nullable TaskState lastSeenTaskState = null;
     private volatile int pollTimeoutsWhileAwaitingFinal = 0;
+    // True once an agent returns without reaching a terminal state (handed work off elsewhere).
+    private volatile boolean asyncAgentPending = false;
+    private volatile int pollTimeoutsWhileAsyncPending = 0;
 
     private static final String ERROR_MSG = "Agent did not return any response";
     private static final int NO_WAIT = -1;
@@ -41,6 +46,12 @@ public class EventConsumer {
     private static final int MAX_AWAITING_FINAL_TIMEOUT_MS = 3000;
     private static final int MAX_POLL_TIMEOUTS_AWAITING_FINAL =
         (MAX_AWAITING_FINAL_TIMEOUT_MS + QUEUE_WAIT_MILLISECONDS - 1) / QUEUE_WAIT_MILLISECONDS;
+    // Bounds otherwise-unbounded polling if an async agent's background work never completes.
+    private static final int MAX_ASYNC_PENDING_TIMEOUT_MS = 60_000;
+    private static final int MAX_POLL_TIMEOUTS_ASYNC_PENDING =
+        (MAX_ASYNC_PENDING_TIMEOUT_MS + QUEUE_WAIT_MILLISECONDS - 1) / QUEUE_WAIT_MILLISECONDS;
+    // Not final so tests can shrink this below the 60s default instead of waiting for it.
+    int maxPollTimeoutsAsyncPending = MAX_POLL_TIMEOUTS_ASYNC_PENDING;
     // Delay between tube.send(finalEvent) and tube.complete() to allow the SSE transport
     // layer to flush the write before the stream-end signal arrives.  Mutiny's internal
     // demand management can call request(1) on the underlying publisher independently of
@@ -83,6 +94,16 @@ public EventConsumer(EventQueue queue, Executor executor) {
         LOGGER.debug("EventConsumer created with queue {}", System.identityHashCode(queue));
     }
 
+    /**
+     * Configures how long to wait for an async agent to reach a terminal state before giving up.
+     *
+     * @param seconds the timeout in seconds
+     */
+    public void setAsyncAgentTimeoutSeconds(int seconds) {
+        this.maxPollTimeoutsAsyncPending =
+            (seconds * 1000 + QUEUE_WAIT_MILLISECONDS - 1) / QUEUE_WAIT_MILLISECONDS;
+    }
+
     public Event consumeOne() throws A2AServerException, EventQueueClosedException {
         EventQueueItem item = queue.dequeueEventItem(NO_WAIT);
         if (item == null) {
@@ -199,12 +220,28 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
                                     LOGGER.debug("Agent completed, awaiting final event (timeout {}/{}), continuing to poll (queue={})",
                                         pollTimeoutsWhileAwaitingFinal, MAX_POLL_TIMEOUTS_AWAITING_FINAL, System.identityHashCode(queue));
                                     pollTimeoutsAfterAgentCompleted = 0; // Reset counter while awaiting final
+                                } else if (asyncAgentPending && isInterruptedState) {
+                                    // Interrupted (non-terminal) state, not async hand-off - timeout doesn't apply.
+                                    pollTimeoutsWhileAsyncPending = 0;
+                                } else if (asyncAgentPending && queueSize == 0 && !awaitingFinal) {
+                                    pollTimeoutsWhileAsyncPending++;
+                                    if (pollTimeoutsWhileAsyncPending >= maxPollTimeoutsAsyncPending) {
+                                        LOGGER.warn("Async agent did not reach a terminal state within {}ms, closing queue (queue={})",
+                                            MAX_ASYNC_PENDING_TIMEOUT_MS, System.identityHashCode(queue));
+                                        queue.close();
+                                        completed = true;
+                                        tube.fail(new InternalError("Agent execution did not complete within the allotted time"));
+                                        return;
+                                    }
+                                } else if (asyncAgentPending && (queueSize > 0 || awaitingFinal)) {
+                                    pollTimeoutsWhileAsyncPending = 0;
                                 }
                                 continue;
                             }
                             // Event received - reset timeout counters
                             pollTimeoutsAfterAgentCompleted = 0;
                             pollTimeoutsWhileAwaitingFinal = 0;
+                            pollTimeoutsWhileAsyncPending = 0;
                             event = item.getEvent();
                             LOGGER.debug("EventConsumer received event: {} (queue={})",
                                 event.getClass().getSimpleName(), System.identityHashCode(queue));
@@ -327,8 +364,14 @@ public EnhancedRunnable.DoneCallback createAgentRunnableDoneCallback() {
                 error = agentRunnable.getError();
                 LOGGER.debug("EventConsumer: Set error field from agent callback");
             } else {
-                agentCompleted = true;
-                LOGGER.debug("EventConsumer: Agent completed successfully, set agentCompleted=true, will close queue after draining");
+                AgentEmitter emitter = agentRunnable.getEmitter();
+                if (emitter != null && !emitter.isTerminalStateReached()) {
+                    asyncAgentPending = true;
+                    LOGGER.debug("EventConsumer: Agent returned without reaching a terminal state, awaiting final event asynchronously");
+                } else {
+                    agentCompleted = true;
+                    LOGGER.debug("EventConsumer: Agent completed successfully, set agentCompleted=true, will close queue after draining");
+                }
             }
         };
     }

server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java

@@ -196,6 +196,7 @@ public class DefaultRequestHandler implements RequestHandler {
     private static final String A2A_BLOCKING_AGENT_TIMEOUT_SECONDS = "a2a.blocking.agent.timeout.seconds";
     private static final String A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS = "a2a.blocking.consumption.timeout.seconds";
     private static final String A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS = "a2a.blocking.reconciliation.timeout.seconds";
+    private static final String A2A_ASYNC_AGENT_TIMEOUT_SECONDS = "a2a.async-agent.timeout.seconds";
     private static final String A2A_REQUEST_CONTEXT_POPULATE_REFERRED_TASKS = "a2a.request-context.populate-referred-tasks";
 
     @Inject
@@ -245,6 +246,18 @@ public class DefaultRequestHandler implements RequestHandler {
      */
     int reconciliationTimeoutSeconds;
 
+    /**
+     * Timeout in seconds to wait for an async agent (one that returns from {@code execute()}
+     * without reaching a terminal {@link AgentEmitter} state) to eventually complete. Bounds
+     * otherwise-unbounded polling if the agent's background work crashes or hangs.
+     * <p>
+     * Property: {@code a2a.async-agent.timeout.seconds}<br>
+     * Default: 60 seconds<br>
+     * Note: Property override requires a configurable {@link A2AConfigProvider} on the classpath
+     * (e.g., MicroProfileConfigProvider in reference implementations).
+     */
+    int asyncAgentTimeoutSeconds;
+
     // Fields set by constructor injection cannot be final. We need a noargs constructor for
     // Jakarta compatibility, and it seems that making fields set by constructor injection
     // final, is not proxyable in all runtimes
@@ -320,6 +333,8 @@ void initConfig() {
                 configProvider.getValue(A2A_BLOCKING_CONSUMPTION_TIMEOUT_SECONDS));
         reconciliationTimeoutSeconds = Integer.parseInt(
                 configProvider.getValue(A2A_BLOCKING_RECONCILIATION_TIMEOUT_SECONDS));
+        asyncAgentTimeoutSeconds = Integer.parseInt(
+                configProvider.getValue(A2A_ASYNC_AGENT_TIMEOUT_SECONDS));
         if (authorizationProviderInstance != null && authorizationProviderInstance.isResolvable()) {
             authorizationProvider = authorizationProviderInstance.get();
         }
@@ -404,6 +419,7 @@ public DefaultRequestHandler build() {
             handler.agentCompletionTimeoutSeconds = 5;
             handler.consumptionCompletionTimeoutSeconds = 2;
             handler.reconciliationTimeoutSeconds = 1;
+            handler.asyncAgentTimeoutSeconds = 60;
             handler.authorizationProvider = authorizationProvider;
             handler.requestContextBuilder =
                     () -> new SimpleRequestContextBuilder(taskStore, populateReferredTasks, authorizationProvider);
@@ -518,6 +534,7 @@ private Task doCancelTask(CancelTaskParams params, ServerCallContext context) th
 
         EventQueue queue = queueManager.createOrTap(task.id());
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
+        consumer.setAsyncAgentTimeoutSeconds(asyncAgentTimeoutSeconds);
 
         // Call agentExecutor.cancel() to enqueue the CANCELED event
         RequestContext cancelRequestContext = requestContextBuilder.get()
@@ -609,6 +626,7 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
 
         // Create consumer BEFORE starting agent - callback is registered inside registerAndExecuteAgentAsync
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
+        consumer.setAsyncAgentTimeoutSeconds(asyncAgentTimeoutSeconds);
 
         EnhancedRunnable producerRunnable = registerAndExecuteAgentAsync(queueTaskId, mss.requestContext, queue, consumer.createAgentRunnableDoneCallback());
 
@@ -667,9 +685,12 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                 // 5. Fetch current task state from TaskStore (includes all consumed & persisted events)
                 LOGGER.debug("DefaultRequestHandler: Entering blocking fire-and-forget handling for task {}", taskId.get());
 
+                boolean isInterruptedState = kind instanceof Task interruptedTask
+                        && interruptedTask.status().state().isInterrupted();
+                boolean isAsync = isAgentAsync(producerRunnable) && !isInterruptedState;
                 try {
-                    // Step 1: Wait for agent to finish (with configurable timeout)
-                    if (agentFuture != null) {
+                    // Step 1: Wait for the agent to finish, unless already handed off asynchronously.
+                    if (agentFuture != null && !isAsync) {
                         try {
                             agentFuture.get(agentCompletionTimeoutSeconds, SECONDS);
                             LOGGER.debug("DefaultRequestHandler: Step 1 - Agent completed for task {}", taskId.get());
@@ -680,11 +701,16 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                         }
                     }
 
-                    // Step 2: Close the queue to signal consumption can complete
-                    // For fire-and-forget tasks, there's no final event, so we need to close the queue
-                    // This allows EventConsumer.consumeAll() to exit
-                    queue.close(false, false);  // graceful close, don't notify parent yet
-                    LOGGER.debug("DefaultRequestHandler: Step 2 - Closed queue for task {} to allow consumption completion", taskId.get());
+                    // Step 2: Close the queue to signal consumption can complete (fire-and-forget
+                    // tasks have no final event otherwise). Re-check isAsync since Step 1 may have
+                    // changed it.
+                    isAsync = isAgentAsync(producerRunnable) && !isInterruptedState;
+                    if (!isAsync) {
+                        queue.close(false, false);  // graceful close, don't notify parent yet
+                        LOGGER.debug("DefaultRequestHandler: Step 2 - Closed queue for task {} to allow consumption completion", taskId.get());
+                    } else {
+                        LOGGER.debug("DefaultRequestHandler: Step 2 - Agent is async, keeping queue open to await final event for task {}", taskId.get());
+                    }
 
                     // Step 3: Wait for consumption to complete (now that queue is closed)
                     if (etai.consumptionFuture() != null) {
@@ -713,10 +739,16 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
                     LOGGER.warn(msg, e.getCause());
                     throw new InternalError(msg);
                 } catch (TimeoutException e) {
-                    // Timeout from consumption future.get() - different from finalization timeout
-                    String msg = String.format("Timeout waiting for task %s consumption", taskId.get());
-                    LOGGER.warn(msg, e);
-                    throw new InternalError(msg);
+                    // For an async agent this is expected (its work legitimately hasn't finished);
+                    // EventConsumer's own fallback timeout still bounds it, so return the task's
+                    // current state instead of failing the call.
+                    if (!isAsync) {
+                        String msg = String.format("Timeout waiting for task %s consumption", taskId.get());
+                        LOGGER.warn(msg, e);
+                        throw new InternalError(msg);
+                    }
+                    LOGGER.debug("DefaultRequestHandler: Step 3 - Async agent for task {} still running after {}s, returning current state",
+                        taskId.get(), consumptionCompletionTimeoutSeconds);
                 }
 
                 // Step 5: Fetch the current task state from TaskStore
@@ -811,6 +843,7 @@ public Flow.Publisher<StreamingEventKind> onMessageSendStream(
 
         // Create consumer BEFORE starting agent - callback is registered inside registerAndExecuteAgentAsync
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
+        consumer.setAsyncAgentTimeoutSeconds(asyncAgentTimeoutSeconds);
 
         EnhancedRunnable producerRunnable = registerAndExecuteAgentAsync(queueTaskId, mss.requestContext, queue, consumer.createAgentRunnableDoneCallback());
 
@@ -1014,6 +1047,7 @@ public Flow.Publisher<StreamingEventKind> onSubscribeToTask(TaskIdParams params,
         // Instead of enqueuing and hoping EventConsumer polls it in time, we prepend it
         // directly to the Publisher stream, ensuring synchronous delivery to subscriber
         EventConsumer consumer = new EventConsumer(queue, eventConsumerExecutor);
+        consumer.setAsyncAgentTimeoutSeconds(asyncAgentTimeoutSeconds);
         Flow.Publisher<EventQueueItem> results = resultAggregator.consumeAndEmit(consumer);
         LOGGER.debug("onSubscribeToTask - prepending initial task snapshot to stream, taskId: {}", params.id());
         return insertingProcessor(
@@ -1054,6 +1088,15 @@ private boolean shouldAddPushInfo(MessageSendParams params) {
         return pushConfigStore != null && params.configuration() != null && params.configuration().taskPushNotificationConfig() != null;
     }
 
+    /**
+     * Returns whether the agent has handed execution off asynchronously: {@code execute()}
+     * returned without its {@link AgentEmitter} reaching a terminal state.
+     */
+    private static boolean isAgentAsync(EnhancedRunnable producerRunnable) {
+        AgentEmitter emitter = producerRunnable.getEmitter();
+        return emitter != null && !emitter.isTerminalStateReached();
+    }
+
     /**
      * Register and execute the agent asynchronously in the agent-executor thread pool.
      *
@@ -1075,6 +1118,7 @@ private EnhancedRunnable registerAndExecuteAgentAsync(String taskId, RequestCont
             public void run() {
                 LOGGER.debug("Agent execution starting for task {}", taskId);
                 AgentEmitter emitter = new AgentEmitter(requestContext, queue);
+                setEmitter(emitter);
                 try {
                     agentExecutor.execute(requestContext, emitter);
                 } catch (A2AError e) {

server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java

@@ -115,6 +115,19 @@ public AgentEmitter(RequestContext context, EventQueue eventQueue) {
         this.contextId = context.getContextId();
     }
 
+    /**
+     * Returns whether a terminal status ({@link #complete}, {@link #fail}, {@link #cancel}, or
+     * {@link #reject}) has been reached.
+     * <p>
+     * If {@link org.a2aproject.sdk.server.agentexecution.AgentExecutor#execute} returns without
+     * this being true, the agent handed work off to another thread.
+     *
+     * @return true if a terminal status has been reached
+     */
+    public boolean isTerminalStateReached() {
+        return terminalStateReached.get();
+    }
+
     /**
      * Updates the task status to the given state with an optional message.
      *

server-common/src/main/resources/META-INF/a2a-defaults.properties

@@ -14,6 +14,11 @@ a2a.blocking.consumption.timeout.seconds=5
 # When in-memory event capture is empty, polls TaskStore with this bounded timeout
 a2a.blocking.reconciliation.timeout.seconds=1
 
+# Timeout for an async agent (execute() returns without reaching a terminal AgentEmitter
+# state, e.g. real work continues on another thread) to eventually complete (seconds)
+# Bounds otherwise-unbounded polling if the agent's background work crashes or hangs
+a2a.async-agent.timeout.seconds=60
+
 # AsyncExecutorProducer - Thread pool configuration
 # Core pool size for async agent execution
 a2a.executor.core-pool-size=5