diff --git a/CHANGELOG.md b/CHANGELOG.md index a9b20b0..02eed2a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## [Unreleased] +- Keep the SPN workbench's notification subscription active across recovery results; + business observers no longer trigger automatic result-based cancellation. - Keep Notification-T subscriptions open independently of task/result terminal states; maintain stream activity tracking and explicit cancellation/resource ownership. - Add independent registry HTTP authentication/TLS policy and orchestration account login for PSOP diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 223c01b..e1578c3 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -237,10 +237,12 @@ acknowledgement() reports actual ACK or failure; timeout is not a synthesized su completion() completes after the stream actually exits. Retain it across task/result terminal statuses and workflow completion. Explicit cancellation, host-agent shutdown, stream exit/error, or the configured idle timeout ends it. -The bundled SPN sample has a separate business policy: `WorkbenchExtensionLifecycle.handleNotification` explicitly -closes its handle after a validated recovery result. This is sample-controlled cancellation, not SDK termination on -Task-T/Notification-T result status. For a continuously subscribed production host, replace that sample callback and -retain the handle across recovery results until an explicit unsubscribe or shutdown. +The bundled SPN `WorkbenchExtensionLifecycle` also keeps the subscription across recovery results. Its injected +notification callback is a business observer, not a result-driven close policy; an observer failure is logged without +cancelling the stream. `SpringWorkbenchExtensionLifecycle.awaitFirstRecovery` only waits for the first recovery result; +it does not unsubscribe. To explicitly unsubscribe one stream, call its handle's `close()` and await `completion()`; +closing the lifecycle shuts down all of its notification resources. Do not close a handle merely because a Task-T or +recovery result reports a terminal state. Task, authorization and notification use separate transports/runtimes/contexts. Neither extension success nor failure is a prerequisite for workflow execution. Authorization controls only the host-defined action associated with that policy. diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 1c96524..4e2143e 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -203,7 +203,7 @@ ExtensionSender.openNotification(agentName, finalContent, (handle, received) -> acknowledgement() 代表真实 ACK 或失败；超时不伪造成功。 close() 请求关闭，completion() 在流实际退出后完成。 任务或通知结果终态不关闭订阅，单次工作流结束也不关闭订阅。有传输活动时持续保持；显式取消、宿主退出、流退出或异常、配置的空闲超时才结束订阅。 -随附 SPN 样例另有业务策略：`WorkbenchExtensionLifecycle.handleNotification` 在校验完整抢通结果后显式关闭自己的 handle。这是样例业务主动取消，不是 SDK 随 Task-T/Notification-T 结果状态自动终止。生产宿主需要持续订阅时，应替换该样例回调，跨抢通结果保留 handle，直到显式取消订阅或宿主退出。 +随附 SPN `WorkbenchExtensionLifecycle` 同样跨抢通结果保留订阅。注入的通知回调是业务观察者，不是结果驱动的关闭策略；观察者异常只记录告警，不取消流。`SpringWorkbenchExtensionLifecycle.awaitFirstRecovery` 仅等待首个抢通结果，不退订。要显式退订一条流，调用对应 handle 的 `close()` 并等待 `completion()`；关闭整个 lifecycle 会释放其全部通知资源。不要仅因 Task-T 或抢通结果出现终态而关闭 handle。 任务、授权、通知使用独立 transport/runtime/context。授权和订阅成功与否均不作为工作流执行前提；授权策略只控制与该策略关联的宿主自定义业务操作。 diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java index a17e237..18116a4 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchExtensionLifecycle.java @@ -303,15 +303,8 @@ public synchronized void close() { private void handleNotification( NotificationSubscription handle, dev.openan.workflow.engine.model.ReceivedMessage received) { - if (dev.openan.workflow.engine.examples.extension.RecoveryNotification.hasCompletedResult( - received)) { - subscriptions.remove(handle.agentName(), handle); - log.info( - "[ExtensionLifecycle] NOTIFICATION_COMPLETE agent={}, contextId={}, action=close-stream", - handle.agentName(), - handle.contextId()); - handle.close(); - } + // Notification-T is workbench-scoped. A recovery result is business data, + // not cancellation of this subscription; keep it for subsequent results. if (notificationCallback != null) { try { notificationCallback.accept(handle, received); diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchNotificationLifecycleTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchNotificationLifecycleTest.java new file mode 100644 index 0000000..f06f2d6 --- /dev/null +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchNotificationLifecycleTest.java @@ -0,0 +1,240 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. You may obtain + * a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ +package dev.openan.workflow.engine.examples.workbench; + +import static org.junit.jupiter.api.Assertions.*; + +import dev.openan.workflow.engine.client.A2ATExtension; +import dev.openan.workflow.engine.client.A2AJavaClientRuntime; +import dev.openan.workflow.engine.client.NotificationSubscription; +import dev.openan.workflow.engine.examples.extension.RecoveryNotification; +import dev.openan.workflow.engine.examples.testsupport.OfflineA2ATLlmClient; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; +import org.a2aproject.sdk.client.ClientEvent; +import org.a2aproject.sdk.client.TaskEvent; +import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.Artifact; +import org.a2aproject.sdk.spec.MessageSendParams; +import org.a2aproject.sdk.spec.Task; +import org.a2aproject.sdk.spec.TaskState; +import org.a2aproject.sdk.spec.TaskStatus; +import org.a2aproject.sdk.spec.TextPart; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** Real lifecycle/transport/handle; only network runtime and content LLM are offline substitutes. */ +@Timeout(30) +class WorkbenchNotificationLifecycleTest { + private String previousLocations; + private String previousEnv; + + @BeforeEach + void configureOfflineSdkAndSingleAgent() { + previousLocations = System.getProperty("A2A_AGENT_CARD_LOCATIONS"); + previousEnv = System.getProperty("a2at.env.path"); + System.setProperty("A2A_AGENT_CARD_LOCATIONS", "classpath:agentcard/spn_domain_agent_city1.json"); + System.setProperty("a2at.env.path", OfflineA2ATLlmClient.envPath()); + OfflineA2ATLlmClient.install(); + } + + @AfterEach + void restoreProperties() { + restore("A2A_AGENT_CARD_LOCATIONS", previousLocations); + restore("a2at.env.path", previousEnv); + } + + private static void restore(String key, String value) { + if (value == null) System.clearProperty(key); + else System.setProperty(key, value); + } + + @Test + void recoveryResultsAndHeartbeatKeepSameSubscriptionUntilExplicitUnsubscribe() throws Exception { + try (Scenario scenario = new Scenario(false)) { + scenario.start(); + scenario.notification.result("result-1", "成功"); + NotificationSubscription handle = scenario.handle.get(); + assertTrue(handle.isActive(), "first completed recovery must not close the subscription"); + long eventCount = handle.heartbeat().eventCount(); + scenario.notification.heartbeat(); + assertTrue(handle.isHealthy(Duration.ofSeconds(5))); + assertEquals(eventCount, handle.heartbeat().eventCount(), "heartbeat is not a business event"); + scenario.notification.result("result-2", "失败"); + assertSame(handle, scenario.handle.get()); + assertTrue(handle.isActive(), "failed recovery result must not close the subscription"); + assertEquals(2, scenario.results.get()); + assertFalse(handle.completion().isDone()); + assertEquals(1, scenario.authorization.closes.get(), "authorization is one-shot"); + assertEquals(0, scenario.notification.closes.get()); + + handle.close(); + handle.completion().get(2, TimeUnit.SECONDS); + assertFalse(handle.isActive()); + } + } + + @Test + void observerFailureDoesNotCancelStreamOrPreventNextResult() throws Exception { + try (Scenario scenario = new Scenario(true)) { + scenario.start(); + scenario.notification.result("result-1", "成功"); + assertTrue(scenario.handle.get().isActive()); + scenario.notification.heartbeat(); + scenario.notification.result("result-2", "失败"); + assertEquals(2, scenario.results.get()); + assertTrue(scenario.handle.get().isActive()); + } + } + + @Test + void hostShutdownClosesStreamAfterRecoveryAndIsIdempotent() throws Exception { + try (Scenario scenario = new Scenario(false)) { + scenario.start(); + scenario.notification.result("result-1", "成功"); + NotificationSubscription handle = scenario.handle.get(); + assertTrue(handle.isActive()); + scenario.lifecycle.close(); + handle.completion().get(2, TimeUnit.SECONDS); + assertFalse(handle.isActive()); + assertFalse(scenario.lifecycle.isActive()); + scenario.lifecycle.close(); + assertEquals(1, scenario.notification.closes.get()); + assertEquals(1, scenario.authorization.closes.get()); + } + } + + private static final class Scenario implements AutoCloseable { + final SimulatedRuntime authorization = new SimulatedRuntime(false); + final SimulatedRuntime notification = new SimulatedRuntime(true); + final AtomicReference handle = new AtomicReference<>(); + final AtomicInteger results = new AtomicInteger(); + final WorkbenchExtensionLifecycle lifecycle; + + Scenario(boolean failFirstObserver) { + AtomicInteger runtimes = new AtomicInteger(); + lifecycle = new WorkbenchExtensionLifecycle(null, true, OfflineA2ATLlmClient.envPath(), + () -> runtimes.getAndIncrement() == 0 ? authorization : notification, + (subscription, event) -> { + handle.set(subscription); + if (RecoveryNotification.hasCompletedResult(event)) { + int count = results.incrementAndGet(); + if (failFirstObserver && count == 1) throw new IllegalStateException("synthetic observer failure"); + } + }, (agentName, card, headers) -> headers.put("Authorization", "Bearer synthetic-notification-test")); + } + + void start() { + lifecycle.start(); + assertTrue(lifecycle.isActive()); + assertNotNull(handle.get()); + assertTrue(handle.get().isActive()); + } + + @Override + public void close() { + lifecycle.close(); + } + } + + private static final class SimulatedRuntime implements A2AJavaClientRuntime { + private static final String RESULT = "### 业务抢通事件\n" + + "1. 业务抢通方案执行状态：已结束\n2. 投诉诊断任务流水号：task-42\n" + + "3. OSS侧事件流水号：event-42\n4. 接入端口名称：port-42\n" + + "5. 是否已授权OMC自动抢通：是\n6. 业务抢通方案名称：隧道调优\n" + + "7. 业务抢通方案详情：执行指定调优策略\n8. 业务抢通方案执行结束时间：2026-08-31T00:00:00Z\n" + + "9. 业务抢通方案执行结果："; + final AtomicInteger closes = new AtomicInteger(); + private final boolean notification; + private final BlockingQueue actions = new LinkedBlockingQueue<>(); + private Consumer sink; + private Runnable activity; + private String contextId; + + SimulatedRuntime(boolean notification) { + this.notification = notification; + } + + @Override + public Iterable sendMessage(AgentCard card, MessageSendParams params, + ClientCallContext callContext, Consumer eventSink, Consumer logSink) { + if (!notification) return List.of(event(TaskState.TASK_STATE_COMPLETED, List.of())); + assertEquals(NOTIFICATION_CHANNEL, callContext.getState().get(CHANNEL_STATE_KEY)); + sink = eventSink; + activity = (Runnable) callContext.getState().get(TRANSPORT_ACTIVITY_STATE_KEY); + contextId = params.message().contextId(); + sink.accept(event(TaskState.TASK_STATE_WORKING, List.of(artifact("subscription", "subscription", "订阅结果：成功")))); + try { + while (!Thread.currentThread().isInterrupted()) actions.take().run(); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } + return List.of(); + } + + void result(String id, String outcome) throws Exception { + submit(() -> sink.accept(event(TaskState.TASK_STATE_WORKING, + List.of(artifact(id, "recovery-result", RESULT + outcome))))); + } + + void heartbeat() throws Exception { + submit(activity); + } + + private void submit(Runnable action) throws Exception { + CompletableFuture delivered = new CompletableFuture<>(); + actions.add(() -> { + try { + action.run(); + delivered.complete(null); + } catch (RuntimeException | Error error) { + delivered.completeExceptionally(error); + throw error; + } + }); + delivered.get(2, TimeUnit.SECONDS); + } + + private static Artifact artifact(String id, String name, String content) { + return Artifact.builder().artifactId(id).name(name).parts(List.of(new TextPart("synthetic event"))) + .metadata(Map.of(A2ATExtension.NOTIFICATION_T.uri(), content)).build(); + } + + private TaskEvent event(TaskState state, List artifacts) { + return new TaskEvent(Task.builder().id("notification-task").contextId(contextId == null ? "auth-context" : contextId) + .status(new TaskStatus(state)).artifacts(artifacts).history(List.of()).metadata(Map.of()).build()); + } + + @Override + public void close() { + closes.incrementAndGet(); + } + } +} diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java index ebba756..9d3a387 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java @@ -110,6 +110,9 @@ public AgentCard agentCard(A2AProperties props, ResourceLoader resourceLoader) t @Bean @ConditionalOnMissingBean public A2AConfigProvider a2aConfigProvider(A2AProperties properties) { + // Kept for SDK components that resolve configuration through CDI; the builder-built + // DefaultRequestHandler never reads it, so requestHandler() applies the blocking timeouts + // directly (see applyBlockingTimeouts). return new A2AConfigProvider() { @Override public String getValue(String key) { @@ -199,8 +202,9 @@ public RequestHandler requestHandler( PushNotificationConfigStore pushStore, MainEventBusProcessor proc, ExecutorService pool, + A2AProperties properties, ObjectProvider authorizationProvider) { - RequestHandler delegate = + DefaultRequestHandler delegate = DefaultRequestHandler.builder() .agentExecutor(executor) .taskStore(store) @@ -210,6 +214,7 @@ public RequestHandler requestHandler( .executor(pool) .eventConsumerExecutor(pool) .build(); + applyBlockingTimeouts(delegate, properties); RequestHandler handler = new SpringCompatibleRequestHandler(delegate); TaskAuthorizationProvider provider = authorizationProvider.getIfAvailable(); if (provider != null) { @@ -252,4 +257,41 @@ public A2AController a2aController( A2ASlashActionAliasController a2aSlashActionAliasController(A2AController delegate) { return new A2ASlashActionAliasController(delegate); } + + /** + * a2a-java's {@code DefaultRequestHandler.Builder.build()} hardcodes blocking-call timeouts + * (agent=5s, consumption=2s, reconciliation=1s) and never consults {@link A2AConfigProvider}; + * only the CDI {@code @PostConstruct} path reads the provider. A builder-assembled handler + * therefore returns a WORKING task snapshot after 5 seconds even though the agent keeps running, + * which silently breaks blocking {@code message:send} for every real agent task (LLM calls, + * multi-step workflows). Apply the configured values directly after the build. + */ + static void applyBlockingTimeouts(DefaultRequestHandler handler, A2AProperties properties) { + try { + setBlockingTimeout( + handler, "agentCompletionTimeoutSeconds", properties.getAgentTimeoutSeconds()); + setBlockingTimeout( + handler, + "consumptionCompletionTimeoutSeconds", + properties.getConsumptionTimeoutSeconds()); + setBlockingTimeout( + handler, "reconciliationTimeoutSeconds", properties.getReconciliationTimeoutSeconds()); + } catch (ReflectiveOperationException e) { + throw new IllegalStateException( + "Failed to apply a2a blocking timeouts; the a2a-java SDK field layout changed?", e); + } + log.info( + "[A2A] Blocking timeouts applied: agent={}s, consumption={}s, reconciliation={}s", + properties.getAgentTimeoutSeconds(), + properties.getConsumptionTimeoutSeconds(), + properties.getReconciliationTimeoutSeconds()); + } + + private static void setBlockingTimeout( + DefaultRequestHandler handler, String fieldName, int seconds) + throws ReflectiveOperationException { + java.lang.reflect.Field field = DefaultRequestHandler.class.getDeclaredField(fieldName); + field.setAccessible(true); + field.setInt(handler, seconds); + } } diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java index 3c61ec1..b2ec1b0 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java @@ -138,7 +138,7 @@ public void shutdown() { public ResponseEntity sendMessage(HttpServletRequest req, @RequestBody String body) { var ctx = buildContext(req); var resp = restHandler.sendMessage(ctx, "", body); - return toResponse(resp); + return toResponse(resp, ctx); } @PostMapping( @@ -289,7 +289,7 @@ public void onComplete() { public ResponseEntity getTask(HttpServletRequest req, @PathVariable("id") String taskId) { var ctx = buildContext(req); var resp = restHandler.getTask(ctx, "", taskId, null); - return toResponse(resp); + return toResponse(resp, ctx); } @GetMapping("${a2at.server.path-prefix}/tasks") @@ -302,9 +302,10 @@ public ResponseEntity listTasks( @RequestParam(required = false) Integer historyLength, @RequestParam(required = false) String statusTimestampAfter, @RequestParam(required = false) Boolean includeArtifacts) { + var ctx = buildContext(req); var resp = restHandler.listTasks( - buildContext(req), + ctx, "", contextId, status, @@ -313,7 +314,7 @@ public ResponseEntity listTasks( historyLength, statusTimestampAfter, includeArtifacts); - return toResponse(resp); + return toResponse(resp, ctx); } @PostMapping("${a2at.server.path-prefix}/tasks/{id}:cancel") @@ -323,7 +324,7 @@ public ResponseEntity cancelTask( @RequestBody(required = false) String body) { var ctx = buildContext(req); var resp = restHandler.cancelTask(ctx, "", body != null ? body : "{}", taskId); - return toResponse(resp); + return toResponse(resp, ctx); } @PostMapping( @@ -415,6 +416,39 @@ private static ResponseEntity toResponse(RestHandler.HTTPRestResponse re .body(response.getBody()); } + /** + * Success responses additionally echo the {@code A2A-Extensions} header with the requested + * extensions that this agent's card declares, so extension-aware clients can verify server + * participation (A2A extension mechanism). Requested extensions the card does not declare are not + * activated and stay out of the list. + */ + private ResponseEntity toResponse( + RestHandler.HTTPRestResponse response, ServerCallContext context) { + String activated = activatedExtensions(context); + ResponseEntity.BodyBuilder builder = + ResponseEntity.status(response.getStatusCode()).contentType(A2A_JSON); + if (!activated.isEmpty()) { + builder.header("A2A-Extensions", activated); + } + return builder.body(response.getBody()); + } + + private String activatedExtensions(ServerCallContext context) { + if (context == null + || context.getRequestedExtensions() == null + || context.getRequestedExtensions().isEmpty() + || agentCard.capabilities() == null + || agentCard.capabilities().extensions() == null) { + return ""; + } + return context.getRequestedExtensions().stream() + .filter( + uri -> + agentCard.capabilities().extensions().stream() + .anyMatch(extension -> uri.equals(extension.uri()))) + .collect(java.util.stream.Collectors.joining(",")); + } + private ServerCallContext buildContext(HttpServletRequest req) { Map state = new LinkedHashMap<>(); Map headers = new LinkedHashMap<>(); diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java index 5a5c2ce..83beacd 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java @@ -45,14 +45,19 @@ public class A2AProperties { */ private boolean slashActionAliasesEnabled = false; - /** Timeout for blocking agent execution. */ - private int agentTimeoutSeconds = 30; + /** + * Timeout for blocking agent execution, in seconds. The a2a-java builder path hardcodes 5 seconds + * and ignores configuration providers; this starter applies the value directly after building the + * handler, so a blocking {@code message:send} waits for the terminal task state. Real agent tasks + * (LLM calls, multi-step workflows) routinely run for minutes. Default: 900. + */ + private int agentTimeoutSeconds = 900; - /** Timeout for consuming and persisting events in blocking calls. */ - private int consumptionTimeoutSeconds = 5; + /** Timeout for consuming and persisting events in blocking calls, in seconds. Default: 960. */ + private int consumptionTimeoutSeconds = 960; - /** Timeout for TaskStore reconciliation polling in blocking calls. */ - private int reconciliationTimeoutSeconds = 1; + /** Timeout for TaskStore reconciliation polling in blocking calls, in seconds. Default: 60. */ + private int reconciliationTimeoutSeconds = 60; /** Core server executor thread count. */ private int executorCoreSize = 8; diff --git a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java index 5200403..4a6c5ad 100644 --- a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java +++ b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java @@ -34,6 +34,7 @@ import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch; import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @@ -127,6 +128,60 @@ void everyTaskRouteUsesConfiguredA2aPathPrefix() throws Exception { assertEquals("text/event-stream", streamMethod.getAnnotation(PostMapping.class).produces()[0]); } + @Test + void sendResponseEchoesRequestedSupportedExtensions() throws Exception { + RestHandler restHandler = mock(RestHandler.class); + RequestHandler requestHandler = mock(RequestHandler.class); + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.capabilities()) + .thenReturn( + AgentCapabilities.builder() + .streaming(true) + .extensions( + List.of( + org.a2aproject.sdk.spec.AgentExtension.builder() + .uri("https://example.org/extensions/demo/v1") + .build())) + .build()); + when(restHandler.sendMessage(any(), any(), any())) + .thenReturn(new RestHandler.HTTPRestResponse(200, "application/a2a+json", "{\"ok\":true}")); + var controller = new A2AController(restHandler, requestHandler, agentCard); + MockMvc mockMvc = + MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + // Requested + supported -> echoed; requested but unsupported -> not activated. + mockMvc + .perform( + post("/a2a/json/message:send") + .header( + "A2A-Extensions", + "https://example.org/extensions/demo/v1," + + "https://example.org/extensions/unknown/v1") + .contentType("application/json") + .content("{}")) + .andExpect(status().isOk()) + .andExpect(header().string("A2A-Extensions", "https://example.org/extensions/demo/v1")); + + // A card without extension declarations activates nothing. + AgentCard plainCard = mock(AgentCard.class); + when(plainCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build()); + var plainController = new A2AController(restHandler, requestHandler, plainCard); + MockMvc plainMockMvc = + MockMvcBuilders.standaloneSetup(plainController) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + plainMockMvc + .perform( + post("/a2a/json/message:send") + .header("A2A-Extensions", "https://example.org/extensions/demo/v1") + .contentType("application/json") + .content("{}")) + .andExpect(status().isOk()) + .andExpect(header().doesNotExist("A2A-Extensions")); + } + @Test void completedSseReleasesPublisherAndSdkEventConsumer() throws Exception { RestHandler restHandler = mock(RestHandler.class); @@ -470,6 +525,54 @@ void blockingTimeoutsComeFromProperties() { assertEquals("3", provider.getValue("a2a.blocking.reconciliation.timeout.seconds")); } + /** + * The builder-built {@code DefaultRequestHandler} never consults {@code A2AConfigProvider} (only + * the CDI {@code @PostConstruct} path does), so the starter must apply the configured timeouts + * directly after {@code build()}. This test exercises that real path: build a handler exactly + * like the autoconfiguration does, apply, then read back the handler fields. + */ + @Test + void blockingTimeoutsAreAppliedToTheBuiltHandler() throws Exception { + var builder = + org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler.builder() + .agentExecutor(mock(org.a2aproject.sdk.server.agentexecution.AgentExecutor.class)) + .taskStore(new org.a2aproject.sdk.server.tasks.InMemoryTaskStore()) + .queueManager(mock(org.a2aproject.sdk.server.events.QueueManager.class)) + .mainEventBusProcessor( + mock(org.a2aproject.sdk.server.events.MainEventBusProcessor.class)) + .executor(java.util.concurrent.Executors.newSingleThreadExecutor()) + .eventConsumerExecutor(java.util.concurrent.Executors.newSingleThreadExecutor()); + var handler = builder.build(); + // a2a-java hardcodes these on the builder path; assert the broken baseline first. + assertEquals(5, readBlockingTimeout(handler, "agentCompletionTimeoutSeconds")); + + A2AProperties properties = new A2AProperties(); + A2AAutoConfiguration.applyBlockingTimeouts(handler, properties); + assertEquals( + properties.getAgentTimeoutSeconds(), + readBlockingTimeout(handler, "agentCompletionTimeoutSeconds")); + assertEquals( + properties.getConsumptionTimeoutSeconds(), + readBlockingTimeout(handler, "consumptionCompletionTimeoutSeconds")); + assertEquals( + properties.getReconciliationTimeoutSeconds(), + readBlockingTimeout(handler, "reconciliationTimeoutSeconds")); + } + + private static int readBlockingTimeout(Object handler, String fieldName) throws Exception { + var field = handler.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + return field.getInt(handler); + } + + @Test + void blockingTimeoutDefaultsSupportLongRunningAgents() { + A2AProperties properties = new A2AProperties(); + assertEquals(900, properties.getAgentTimeoutSeconds()); + assertEquals(960, properties.getConsumptionTimeoutSeconds()); + assertEquals(60, properties.getReconciliationTimeoutSeconds()); + } + @Test void executorUsesConfiguredBounds() { A2AProperties properties = new A2AProperties();