feat(engine): make fallback task poll interval configurable (default 20s)

The fallback task-polling interval (used when an SSE stream ends before
a terminal state) was hardcoded at 250ms, causing ~1000 getTask polls
and log spam during a 138-second task execution observed in field logs.

Add taskPollIntervalMillis to WorkflowEngineClientConfig (default 20000ms
= 20 seconds). Polling only triggers when a streaming response is
interrupted before the task reaches a terminal state (e.g. a gateway
closes the SSE channel mid-stream); while SSE is alive (heartbeats still
arriving), polling does not start. Tests that exercise the polling path
use 100ms for fast feedback.

Signed-off-by: Zhoujie628 <zhoujie628@users.noreply.github.com>

Author: Zhoujie628

docs/en/API_REFERENCE.md

@@ -150,6 +150,7 @@ Builder-based configuration for the workflow engine client.
 | `crlPath`                       | `String`       | null    | X.509 CRL for HTTP/JSON-RPC; the default gRPC runtime rejects this unsupported combination                                                              |
 | `sendTimeoutSeconds`            | `long`         | `600`   | SSE stream timeout (10 min default)                                                                                                                     |
 | `notificationAckTimeoutSeconds` | `long`         | `300`   | Wait for the first Notification-T ACK/event (five minutes by default)                                                                                   |
+| `taskPollIntervalMillis`        | `long`         | `20000` | Fallback task polling interval when an SSE stream ends before a terminal state (e.g. a gateway closes the channel mid-stream); 20 seconds by default. While SSE is alive (heartbeats still arriving), polling does not start |
 | `sendExecutorCoreSize`          | `int`          | `4`     | Send executor core threads                                                                                                                              |
 | `sendExecutorMaxSize`           | `int`          | `16`    | Send executor maximum threads                                                                                                                           |
 | `sendExecutorQueueCapacity`     | `int`          | `256`   | Bounded send executor queue capacity                                                                                                                    |

docs/zh/API_REFERENCE.md

@@ -137,6 +137,7 @@ handle 再开始 I/O，监听器直接收到 handle 和完整 ReceivedMessage。
 | `crlPath`                       | `String`       | null   | HTTP/JSON-RPC 的 X.509 CRL 路径；默认 gRPC runtime 暂不支持并会拒绝启动                                   |
 | `sendTimeoutSeconds`            | `long`         | `600`  | SSE 流超时（默认 10 分钟）                                                                                |
 | `notificationAckTimeoutSeconds` | `long`         | `300`  | Notification-T 首个 ACK/事件等待时间（默认 5 分钟）                                                       |
+| `taskPollIntervalMillis`        | `long`         | `20000`| SSE 流中途断开（如网关空闲超时掐流）后的兜底任务轮询间隔（默认 20 秒）。SSE 流存活期间（心跳仍在推送时）不会启动轮询 |
 | `sendExecutorCoreSize`          | `int`          | `4`    | 发送线程池核心线程数                                                                                      |
 | `sendExecutorMaxSize`           | `int`          | `16`   | 发送线程池最大线程数                                                                                      |
 | `sendExecutorQueueCapacity`     | `int`          | `256`  | 发送线程池有界队列容量                                                                                    |

samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java

@@ -171,7 +171,7 @@ void setUp() throws Exception {
             null,
             WorkflowEngineClientConfig.builder().sslVerify(false).build());
 
-    client = new DefaultWorkflowEngineClient(transport);
+    client = new DefaultWorkflowEngineClient(transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build());
     client.setControlPoint(new WorkbenchControlPoint(sdkEnvPath));
   }
 

workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java

@@ -57,6 +57,7 @@ public class DefaultWorkflowEngineClient implements WorkflowEngineClient, AutoCl
   private final A2ATransport transport;
   private final int maxNegotiationExchanges;
   private final boolean closeTransportOnClose;
+  private final long taskPollIntervalMillis;
   private final AtomicBoolean closed = new AtomicBoolean();
   private final Set<Invocation> invocations = java.util.concurrent.ConcurrentHashMap.newKeySet();
   private final ScheduledExecutorService timeoutScheduler =
@@ -76,15 +77,27 @@ public DefaultWorkflowEngineClient(A2ATransport transport) {
 
   /** Uses the configured resource exchange budget, not an SDK protocol round counter. */
   public DefaultWorkflowEngineClient(A2ATransport transport, WorkflowEngineClientConfig config) {
-    this(transport, config.getMaxNegotiationExchanges(), false);
+    this(transport, config.getMaxNegotiationExchanges(), false, config.getTaskPollIntervalMillis());
   }
 
   private DefaultWorkflowEngineClient(A2ATransport transport, int maxExchanges, boolean owning) {
+    this(
+        transport,
+        maxExchanges,
+        owning,
+        WorkflowEngineClientConfig.DEFAULT_TASK_POLL_INTERVAL_MILLIS);
+  }
+
+  private DefaultWorkflowEngineClient(
+      A2ATransport transport, int maxExchanges, boolean owning, long taskPollIntervalMillis) {
     this.transport = Objects.requireNonNull(transport, "transport");
     if (maxExchanges < 1)
       throw new IllegalArgumentException("Negotiation exchange budget must be positive");
+    if (taskPollIntervalMillis < 100)
+      throw new IllegalArgumentException("Task poll interval must be at least 100ms");
     this.maxNegotiationExchanges = maxExchanges;
     this.closeTransportOnClose = owning;
+    this.taskPollIntervalMillis = taskPollIntervalMillis;
   }
 
   /** Creates a facade owning the supplied transport. */
@@ -95,7 +108,8 @@ public static DefaultWorkflowEngineClient owning(A2ATransport transport) {
   /** Creates an owning facade with explicit resource settings. */
   public static DefaultWorkflowEngineClient owning(
       A2ATransport transport, WorkflowEngineClientConfig config) {
-    return new DefaultWorkflowEngineClient(transport, config.getMaxNegotiationExchanges(), true);
+    return new DefaultWorkflowEngineClient(
+        transport, config.getMaxNegotiationExchanges(), true, config.getTaskPollIntervalMillis());
   }
 
   private static ReceivedMessage negotiationResponse(SendMessageResult result) {
@@ -580,7 +594,7 @@ private CompletableFuture<Void> delayBeforeTaskPoll(Invocation invocation) {
                 pending.completion.complete(null);
               }
             },
-            250,
+            taskPollIntervalMillis,
             TimeUnit.MILLISECONDS);
     if (pending.completion.isDone()) pending.scheduled.cancel(false);
     return pending.completion;

workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java

@@ -38,6 +38,16 @@ public class WorkflowEngineClientConfig {
   /** Default wait for the first Notification-T acknowledgement: five minutes. */
   public static final long DEFAULT_NOTIFICATION_ACK_TIMEOUT_SECONDS = 300;
 
+  /**
+   * Default interval for fallback task polling (when an SSE stream ends before a terminal state):
+   * 20 seconds. This polling only triggers when a streaming response is interrupted before the task
+   * reaches a terminal state (e.g. a gateway closes the SSE channel mid-stream). While SSE
+   * is alive (heartbeats still arriving), polling does not start; it only begins after the stream
+   * returns a non-terminal state. The 20-second interval avoids hammering the remote endpoint
+   * during long-running agent executions.
+   */
+  public static final long DEFAULT_TASK_POLL_INTERVAL_MILLIS = 20_000;
+
   private final boolean sslVerify;
   private final String caCertsPath;
   private final String clientCertPath;
@@ -55,6 +65,7 @@ public class WorkflowEngineClientConfig {
   private final Map<String, Map<String, Map<String, Object>>> credentialsConfig;
   private final int maxNegotiationExchanges;
   private final String preferredProtocol;
+  private final long taskPollIntervalMillis;
 
   private WorkflowEngineClientConfig(Builder b) {
     this.sslVerify = b.sslVerify;
@@ -75,6 +86,7 @@ private WorkflowEngineClientConfig(Builder b) {
         b.credentialsConfig != null ? copyCredentials(b.credentialsConfig) : null;
     this.maxNegotiationExchanges = b.maxNegotiationExchanges;
     this.preferredProtocol = b.preferredProtocol;
+    this.taskPollIntervalMillis = b.taskPollIntervalMillis;
   }
 
   private static Map<String, Map<String, Map<String, Object>>> copyCredentials(
@@ -136,6 +148,7 @@ public static class Builder {
     private Map<String, Map<String, Map<String, Object>>> credentialsConfig = null;
     private int maxNegotiationExchanges = 3;
     private String preferredProtocol = null;
+    private long taskPollIntervalMillis = DEFAULT_TASK_POLL_INTERVAL_MILLIS;
 
     public Builder sslVerify(boolean v) {
       this.sslVerify = v;
@@ -223,6 +236,15 @@ public Builder preferredProtocol(String v) {
       return this;
     }
 
+    /**
+     * Interval (milliseconds) for fallback task polling when the SSE stream ends before a terminal
+     * state. Defaults to {@value #DEFAULT_TASK_POLL_INTERVAL_MILLIS}ms (20 seconds).
+     */
+    public Builder taskPollIntervalMillis(long v) {
+      this.taskPollIntervalMillis = v;
+      return this;
+    }
+
     public WorkflowEngineClientConfig build() {
       if (sendTimeoutSeconds <= 0 || notificationAckTimeoutSeconds <= 0) {
         throw new IllegalArgumentException("Timeouts must be positive");

workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java

@@ -279,7 +279,8 @@ public ListTasksResult listTasks(
             .build();
 
     try (A2ATransport transport = new A2ATransport(List.of(card), runtime, config)) {
-      new DefaultWorkflowEngineClient(transport)
+      new DefaultWorkflowEngineClient(
+              transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build())
           .listTasks(card.name(), ListTasksParams.builder().pageSize(10).build())
           .join();
     }
@@ -353,7 +354,9 @@ void borrowedClientDoesNotCloseCallerOwnedTransport() throws Exception {
             new CountingRuntime(closeCalls),
             WorkflowEngineClientConfig.builder().build());
 
-    new DefaultWorkflowEngineClient(transport).close();
+    new DefaultWorkflowEngineClient(
+            transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build())
+        .close();
     assertEquals(0, closeCalls.get());
 
     transport.close();
@@ -393,7 +396,10 @@ void finalStructuredContentNeedsNoContentSdk() throws Exception {
               List.of(new org.a2aproject.sdk.spec.DataPart(Map.of("any", List.of(1, 2)))),
               Map.of("contextId", "business-only"),
               java.util.Set.of("urn:optional:custom"));
-      new DefaultWorkflowEngineClient(transport).sendTask("Test Agent", content).join();
+      new DefaultWorkflowEngineClient(
+              transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build())
+          .sendTask("Test Agent", content)
+          .join();
       assertEquals(content.parts(), captured.get().message().parts());
       assertEquals(content.metadata(), captured.get().message().metadata());
       assertFalse("business-only".equals(captured.get().message().contextId()));