diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index 663fef6..8198b6e 100644 --- a/docs/en/API_REFERENCE.md +++ b/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 | diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 9a6a2e7..982ad3f 100644 --- a/docs/zh/API_REFERENCE.md +++ b/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` | 发送线程池有界队列容量 | diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java index 0de706a..297a760 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/server/EmbeddedA2AServerTest.java +++ b/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)); } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java index 68f9e64..0eb90de 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java +++ b/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 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 delayBeforeTaskPoll(Invocation invocation) { pending.completion.complete(null); } }, - 250, + taskPollIntervalMillis, TimeUnit.MILLISECONDS); if (pending.completion.isDone()) pending.scheduled.cancel(false); return pending.completion; diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java index 9a2ec31..9b41b7e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java +++ b/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>> 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>> copyCredentials( @@ -136,6 +148,7 @@ public static class Builder { private Map>> 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"); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java index 46b20cb..1f46f10 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java +++ b/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())); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java index fb1411b..4b55c0a 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java @@ -165,7 +165,9 @@ void standaloneTaskUsesFreshContextAndConfiguredInteractionLoop() throws Excepti MessageContent.text("task-done"))); }), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); assertEquals( "TASK_STATE_COMPLETED", @@ -197,7 +199,9 @@ void standaloneTaskAcceptsPerCallNegotiationStrategy() throws Exception { ? negotiation("standalone", 1, NegotiationPerformative.PROPOSE) : MessageContent.text("done")))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); var result = client @@ -252,7 +256,9 @@ public void close() {} }; try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); assertThrows( CompletionException.class, @@ -338,7 +344,9 @@ public void close() {} }; try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); CompletionException error = assertThrows( @@ -363,7 +371,9 @@ void standaloneTaskValidatesTaskTExtensionOnlyWhenActivated() throws Exception { params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); }), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.sendTask("test", MessageContent.text("plain task")).join(); MessageContent taskT = @@ -392,7 +402,9 @@ void standaloneTaskRejectsNegotiationActivationNotDeclaredByTarget() throws Exce params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); }), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); MessageContent content = new MessageContent( List.of(new TextPart("negotiation-capable task")), @@ -438,7 +450,9 @@ public void close() {} }; try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); CompletionException error = assertThrows( @@ -472,7 +486,9 @@ void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception { AtomicReference seen = new AtomicReference<>(); try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -510,7 +526,9 @@ void terminalStateNeverRestartsOldNegotiation() throws Exception { TaskState.TASK_STATE_COMPLETED, negotiation("old", 1, NegotiationPerformative.PROPOSE)))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -545,7 +563,10 @@ void rejectsWrongContextNewProposeAndMissingCallbackWithoutSendingAgain() throws negotiation("city1", 1, NegotiationPerformative.PROPOSE))); }), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, + WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation(q -> CompletableFuture.completedFuture(reply)) @@ -571,7 +592,9 @@ void plainInputRequiredIsExplicitlyUnsupported() throws Exception { TaskState.TASK_STATE_INPUT_REQUIRED, MessageContent.text("question")))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); assertThrows( CompletionException.class, () -> client.sendTask("test", negotiationTask("start")).join()); @@ -593,7 +616,9 @@ params, negotiation("pre-task", 1, 5, NegotiationPerformative.PROPOSE)) }); try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -643,7 +668,9 @@ void tasklessRoundLoopAdvancesRoundsAndCompletesWithTaskAnchoredResult() throws }); try (var transport = new A2ATransport(List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -681,7 +708,9 @@ void tasklessProposeWithoutHostActivationFailsLoudly() throws Exception { p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE)); }), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); CompletionException error = assertThrows( CompletionException.class, @@ -703,7 +732,9 @@ void tasklessProposeWithMissingNegotiationContextFailsLoudly() throws Exception List.of(card()), runtime(p -> messageResponse(p, malformed)), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); CompletionException error = assertThrows( CompletionException.class, @@ -721,7 +752,9 @@ void duplicateTasklessProposeFailsExplicitlyWithoutRepeatingTheCallback() throws runtime( p -> messageResponse(p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -754,7 +787,9 @@ void tasklessMessageWithoutNegotiationMetadataRemainsNormalResponse() throws Exc List.of(card()), runtime(p -> messageResponse(p, MessageContent.text("plain final"))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -786,7 +821,9 @@ void abortAcknowledgementIsNotDiagnosisSuccessAndPreservesRemoteState() throws E : TaskState.TASK_STATE_COMPLETED, negotiation("city1", 1, NegotiationPerformative.PROPOSE)))), WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -844,7 +881,10 @@ public void close() {} try (var transport = new A2ATransport( List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, + WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -887,7 +927,10 @@ void canceledOrTimedOutNegotiationNeverSendsLateReply() throws Exception { negotiation("city1", 1, NegotiationPerformative.PROPOSE))); }), WorkflowEngineClientConfig.builder().sendTimeoutSeconds(1).build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, + WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); client.setControlPoint( ControlPoint.builder() .onNegotiation( @@ -917,14 +960,18 @@ void closingNonOwningFacadePreventsNewWorkButDoesNotCloseSharedTransport() throw List.of( response(p, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done")))), WorkflowEngineClientConfig.builder().build())) { - var closedClient = new DefaultWorkflowEngineClient(transport); + var closedClient = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); closedClient.close(); assertThrows( CompletionException.class, () -> closedClient.sendTask("test", MessageContent.text("no")).join()); assertEquals( "TASK_STATE_COMPLETED", - new DefaultWorkflowEngineClient(transport) + new DefaultWorkflowEngineClient( + transport, + WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()) .sendTask("test", MessageContent.text("yes")) .join() .getTaskState()); @@ -950,7 +997,9 @@ void concurrentTasksKeepInvocationEventCallbacksIsolated() throws Exception { try (var transport = new A2ATransport( List.of(card()), concurrentRuntime, WorkflowEngineClientConfig.builder().build())) { - var client = new DefaultWorkflowEngineClient(transport); + var client = + new DefaultWorkflowEngineClient( + transport, WorkflowEngineClientConfig.builder().taskPollIntervalMillis(100).build()); List firstEvents = new CopyOnWriteArrayList<>(); List secondEvents = new CopyOnWriteArrayList<>(); List firstRequests = new CopyOnWriteArrayList<>();