fix(a2a): apply configured blocking timeouts, echo A2A-Extensions responses, notification subscription lifecycle

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

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.

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。授权和订阅成功与否均不作为工作流执行前提；授权策略只控制与该策略关联的宿主自定义业务操作。
 

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);

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<NotificationSubscription> 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<Runnable> actions = new LinkedBlockingQueue<>();
+    private Consumer<ClientEvent> sink;
+    private Runnable activity;
+    private String contextId;
+
+    SimulatedRuntime(boolean notification) {
+      this.notification = notification;
+    }
+
+    @Override
+    public Iterable<ClientEvent> sendMessage(AgentCard card, MessageSendParams params,
+        ClientCallContext callContext, Consumer<ClientEvent> eventSink, Consumer<String> 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<Void> 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<Artifact> 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();
+    }
+  }
+}

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<TaskAuthorizationProvider> 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);
+  }
 }