diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md index 9702fd10e..b80227599 100644 --- a/docs/content/dev/server.md +++ b/docs/content/dev/server.md @@ -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. diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index ebda76a1c..aa99eab0a 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -153,9 +153,14 @@ *
  • Blocking (configuration.blocking=true): Client waits for first event or final task state
  • *
  • Streaming: Client receives events as they arrive via reactive streams
  • *
  • Both modes support fire-and-forget (agent continues after client disconnect)
  • - *
  • Configurable timeouts via {@code a2a.blocking.agent.timeout.seconds}, + *
  • 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}
  • + *
  • 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.
  • * * *

    CDI Dependencies

    @@ -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; @@ -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; @@ -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; diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 77f4bb9b4..eac9f77a4 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -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(); @@ -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 callerQueue = new AtomicReference<>(); useQueueManager(new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) { @Override @@ -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) {