Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/content/dev/server.md
Original file line number Diff line number Diff line change
Expand Up @@ -295,4 +295,6 @@ class A2AServerConfiguration {

Register the remaining dependencies, including an `a2aInternalExecutor` bean and a `TaskAuthorizationProvider`, as Spring beans. Keep the cached event consumer executor separate from the agent executor: blocking event polling can exhaust a bounded pool under load.

Manually built handlers use blocking timeouts of 30 seconds for agent execution, 5 seconds for event consumption, and 1 second for TaskStore reconciliation. Override these with `agentCompletionTimeoutSeconds(int)`, `consumptionCompletionTimeoutSeconds(int)`, and `reconciliationTimeoutSeconds(int)` on the builder. The `a2a.blocking.*` configuration properties apply to CDI-managed handlers; manual wiring uses the builder methods.

See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration.
Original file line number Diff line number Diff line change
Expand Up @@ -153,9 +153,14 @@
* <li><b>Blocking (configuration.blocking=true):</b> Client waits for first event or final task state</li>
* <li><b>Streaming:</b> Client receives events as they arrive via reactive streams</li>
* <li>Both modes support fire-and-forget (agent continues after client disconnect)</li>
* <li>Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
* <li>For CDI-managed handlers, configurable timeouts via {@code a2a.blocking.agent.timeout.seconds},
* {@code a2a.blocking.consumption.timeout.seconds}, and
* {@code a2a.blocking.reconciliation.timeout.seconds}</li>
* <li>For manually built handlers, configure timeouts with
* {@link Builder#agentCompletionTimeoutSeconds(int)},
* {@link Builder#consumptionCompletionTimeoutSeconds(int)}, and
* {@link Builder#reconciliationTimeoutSeconds(int)}.
* Both paths default to 30, 5, and 1 seconds, respectively.</li>
* </ul>
*
* <h2>CDI Dependencies</h2>
Expand Down Expand Up @@ -405,6 +410,9 @@ public static class Builder {
private boolean authorizationRequired = true;
private boolean populateReferredTasks;
private boolean pushNotificationsEnabled = true;
private int agentCompletionTimeoutSeconds = 30;
private int consumptionCompletionTimeoutSeconds = 5;
private int reconciliationTimeoutSeconds = 1;

public Builder agentExecutor(AgentExecutor agentExecutor) {
this.agentExecutor = agentExecutor;
Expand Down Expand Up @@ -441,6 +449,51 @@ public Builder eventConsumerExecutor(Executor eventConsumerExecutor) {
return this;
}

/**
* Sets how long a blocking call waits for agent execution. Defaults to 30 seconds.
*
* @param seconds timeout in seconds; zero skips waiting
* @return this builder for chaining
* @throws IllegalArgumentException if the timeout is negative
*/
public Builder agentCompletionTimeoutSeconds(int seconds) {
if (seconds < 0) {
throw new IllegalArgumentException("Agent completion timeout must not be negative");
}
this.agentCompletionTimeoutSeconds = seconds;
return this;
}

/**
* Sets how long a blocking call waits for event consumption. Defaults to 5 seconds.
*
* @param seconds timeout in seconds; zero skips waiting
* @return this builder for chaining
* @throws IllegalArgumentException if the timeout is negative
*/
public Builder consumptionCompletionTimeoutSeconds(int seconds) {
if (seconds < 0) {
throw new IllegalArgumentException("Consumption completion timeout must not be negative");
}
this.consumptionCompletionTimeoutSeconds = seconds;
return this;
}

/**
* Sets how long TaskStore reconciliation polls in blocking calls. Defaults to 1 second.
*
* @param seconds timeout in seconds; zero disables reconciliation polling
* @return this builder for chaining
* @throws IllegalArgumentException if the timeout is negative
*/
public Builder reconciliationTimeoutSeconds(int seconds) {
if (seconds < 0) {
throw new IllegalArgumentException("Reconciliation timeout must not be negative");
}
this.reconciliationTimeoutSeconds = seconds;
return this;
}

public Builder authorizationProvider(@Nullable TaskAuthorizationProvider authorizationProvider) {
this.authorizationProvider = authorizationProvider;
return this;
Expand Down Expand Up @@ -490,9 +543,9 @@ public DefaultRequestHandler build() {
DefaultRequestHandler handler =
new DefaultRequestHandler(agentExecutor, taskStore, queueManager, pushConfigStore,
mainEventBusProcessor, executor, eventConsumerExecutor);
handler.agentCompletionTimeoutSeconds = 5;
handler.consumptionCompletionTimeoutSeconds = 2;
handler.reconciliationTimeoutSeconds = 1;
handler.agentCompletionTimeoutSeconds = agentCompletionTimeoutSeconds;
handler.consumptionCompletionTimeoutSeconds = consumptionCompletionTimeoutSeconds;
handler.reconciliationTimeoutSeconds = reconciliationTimeoutSeconds;
handler.authorizationProvider = authorizationProvider;
handler.agentExecutorRouter = agentExecutorRouter;
handler.authorizationRequired = authorizationRequired;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -203,6 +203,44 @@ void testInitConfigReadsBlockingTimeouts() {
assertEquals(7, handler.reconciliationTimeoutSeconds);
}

@Test
void testBuilderUsesDocumentedBlockingTimeoutDefaults() {
DefaultRequestHandler handler = (DefaultRequestHandler) requestHandler;

assertEquals(30, handler.agentCompletionTimeoutSeconds);
assertEquals(5, handler.consumptionCompletionTimeoutSeconds);
assertEquals(1, handler.reconciliationTimeoutSeconds);
}

@ParameterizedTest
@CsvSource({"41,8,3", "0,0,0"})
void testBuilderOverridesBlockingTimeouts(int agentTimeout, int consumptionTimeout, int reconciliationTimeout) {
DefaultRequestHandler handler = DefaultRequestHandler.builder()
.agentExecutor(executor)
.taskStore(taskStore)
.queueManager(queueManager)
.mainEventBusProcessor(mainEventBusProcessor)
.executor(internalExecutor)
.eventConsumerExecutor(internalExecutor)
.agentCompletionTimeoutSeconds(agentTimeout)
.consumptionCompletionTimeoutSeconds(consumptionTimeout)
.reconciliationTimeoutSeconds(reconciliationTimeout)
.build();

assertEquals(agentTimeout, handler.agentCompletionTimeoutSeconds);
assertEquals(consumptionTimeout, handler.consumptionCompletionTimeoutSeconds);
assertEquals(reconciliationTimeout, handler.reconciliationTimeoutSeconds);
}

@Test
void testBuilderRejectsNegativeBlockingTimeouts() {
DefaultRequestHandler.Builder builder = DefaultRequestHandler.builder();

assertThrows(IllegalArgumentException.class, () -> builder.agentCompletionTimeoutSeconds(-1));
assertThrows(IllegalArgumentException.class, () -> builder.consumptionCompletionTimeoutSeconds(-1));
assertThrows(IllegalArgumentException.class, () -> builder.reconciliationTimeoutSeconds(-1));
}

@Test
void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception {
InMemoryTaskStore manualTaskStore = new InMemoryTaskStore();
Expand Down Expand Up @@ -1758,8 +1796,9 @@ public void cancel(RequestContext context, AgentEmitter emitter) {
* must still reach the queue, which was told to expect it; otherwise the consumer waits out its
* grace period for it and the call fails or returns late.
*/
@Test
void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception {
@ParameterizedTest
@CsvSource({"true", "false"})
void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate(boolean omitConfiguration) throws Exception {
AtomicReference<EventQueue> callerQueue = new AtomicReference<>();
useQueueManager(new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) {
@Override
Expand All @@ -1772,15 +1811,37 @@ public EventQueue createOrTap(String taskId) {
AtomicBoolean callerQueueClosedInTime = new AtomicBoolean();
holdEventsAfterWorkingUntilClosed(callerQueue, callerQueueClosedInTime);
agentExecutorExecute = (context, emitter) -> {
emitter.submit();
emitter.startWork();
emitter.addArtifact(List.of(new TextPart("answer")), "answer-id", "answer", null);
emitter.complete();
};

EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT);
MessageSendParams params = blockingParams("msg-blocking-final-event");
if (omitConfiguration) {
params = MessageSendParams.builder().message(params.message()).build();
}
EventKind result = requestHandler.onMessageSend(params, NULL_CONTEXT);

assertTrue(callerQueueClosedInTime.get(), "the caller's queue was not closed while the final event was held");
Task task = assertInstanceOf(Task.class, result);
assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state());
assertEquals(1, task.artifacts().size());
assertEquals("answer-id", task.artifacts().get(0).artifactId());
assertEquals("answer", assertInstanceOf(TextPart.class, task.artifacts().get(0).parts().get(0)).text());
}

@Test
void blockingSendWithoutFinalEventReturnsWorkingTaskAndAllowsResubscription() throws Exception {
agentExecutorExecute = (context, emitter) -> emitter.startWork();

EventKind result = requestHandler.onMessageSend(blockingParams("msg-fire-and-forget"), NULL_CONTEXT);

Task task = assertInstanceOf(Task.class, result);
assertEquals(TaskState.TASK_STATE_WORKING, task.status().state());
EventQueue resubscribedQueue = queueManager.tap(task.id());
assertNotNull(resubscribedQueue);
resubscribedQueue.close();
}

private void useQueueManager(InMemoryQueueManager manager) {
Expand Down
Loading