From 5034ee545b61c19f5c52a0b90bec6d05c0f09285 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Sun, 13 Sep 2026 17:06:05 +0800 Subject: [PATCH 1/2] fix(engine): harden concurrent protocol lifecycles Signed-off-by: Zhoujie628 --- pom.xml | 9 ++ .../examples/demo/A2ATProtocolCases.java | 4 +- .../engine/examples/demo/HostQuickStart.java | 2 +- .../WorkbenchExtensionLifecycle.java | 4 +- .../workbench/WorkbenchOrchestrator.java | 3 - .../engine/examples/SpnCrossCityE2ETest.java | 5 +- .../examples/demo/HostQuickStartTest.java | 3 +- workflow-engine/pom.xml | 52 ++++++- .../engine/client/A2AJavaClientRuntime.java | 1 + .../workflow/engine/client/A2ATransport.java | 27 +++- .../engine/client/AgentAuthManager.java | 2 +- .../client/DefaultA2AJavaClientRuntime.java | 17 +++ .../engine/client/DefaultExtensionSender.java | 33 ++++- .../client/DefaultWorkflowEngineClient.java | 138 ++++++++++++++---- .../client/NotificationSubscription.java | 31 +++- .../engine/client/ObservedHttpClient.java | 2 + .../client/TransportActivityMonitor.java | 57 ++++++++ .../engine/client/WorkflowEngineClient.java | 13 ++ .../client/WorkflowEngineClientConfig.java | 8 +- .../engine/core/WorkflowExecutor.java | 5 +- .../workflow/engine/registry/LoadPsop.java | 76 +++++++++- .../workflow/engine/runner/ExecutePsop.java | 10 ++ ...ltWorkflowEngineClientNegotiationTest.java | 62 ++++++++ .../engine/client/ExtensionSenderTest.java | 52 +++++++ .../client/NotificationSubscriptionTest.java | 6 + .../engine/client/ObservedHttpClientTest.java | 40 +++++ .../WorkflowEngineClientConfigTest.java | 17 +++ .../engine/core/WorkflowExecutorTest.java | 15 +- .../engine/registry/LoadPsopSearchTest.java | 50 +++++++ .../engine/runner/ExecutePsopTest.java | 18 ++- 30 files changed, 689 insertions(+), 73 deletions(-) create mode 100644 workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportActivityMonitor.java diff --git a/pom.xml b/pom.xml index 57388365..cec6a13d 100644 --- a/pom.xml +++ b/pom.xml @@ -60,6 +60,7 @@ 2.22.0 2.0.18 5.10.2 + 1.81.0 2.24.3 3.3.10 0.11.0 @@ -71,6 +72,14 @@ + + + org.junit + junit-bom + ${junit.version} + pom + import + org.a2aproject.sdk a2a-java-sdk-bom diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java index dd964959..8f6543dc 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/A2ATProtocolCases.java @@ -174,8 +174,8 @@ public void onEvent(String type, Map data) { } }); notificationControlClient = new DefaultWorkflowEngineClient(notificationTransport); - authorizationSender = new DefaultExtensionSender(authorizationTransport); - notificationSender = new DefaultExtensionSender(notificationTransport); + authorizationSender = DefaultExtensionSender.nonOwning(authorizationTransport); + notificationSender = DefaultExtensionSender.nonOwning(notificationTransport); } private void teardown() { diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/HostQuickStart.java b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/HostQuickStart.java index 949e58bc..37c1dfbb 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/HostQuickStart.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/HostQuickStart.java @@ -80,7 +80,7 @@ public static ExecutionResult execute(Workflow workflow, List cards, .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Host review required"))) .build(); - CompletableFuture execution = ExecutePsop.builder().psop(workflow).agentCards(cards) + CompletableFuture execution = ExecutePsop.builder().psop(workflow) .engineClient(client).controlPoint(callbacks).runtimeIntent("Diagnose the supplied issue").execute(); try { return execution.get(10, TimeUnit.MINUTES); 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 8012e1ac..a17e2378 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 @@ -200,8 +200,8 @@ public synchronized void start() { List opened = new ExtensionPrePositioner(a2atEnvPath) .prePosition( - new DefaultExtensionSender(authorizationTransport), - new DefaultExtensionSender(notificationCandidate), + DefaultExtensionSender.nonOwning(authorizationTransport), + DefaultExtensionSender.nonOwning(notificationCandidate), agentCards, this::handleNotification, subscription -> subscriptions.put(subscription.agentName(), subscription)); diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java index d5d55c00..e9ea0d5d 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java @@ -314,13 +314,10 @@ String run(String messageText, boolean demoNegotiationEnabled, ExecutionCancella CompletableFuture execution = ExecutePsop.builder() .psop(workflow) - .agentCards(agentCards) .controlPoint(controlPoint) .engineClient(engineClient) .runtimeIntent(messageText) .lang("zh") - .sslVerify(sslVerify) - .credentialsConfigPath(credentialsPath) .eventCallback(createLogCallback(transport.getContextId())) .onFinish( (r, events) -> { diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/SpnCrossCityE2ETest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/SpnCrossCityE2ETest.java index 0099df6a..b514e01e 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/SpnCrossCityE2ETest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/SpnCrossCityE2ETest.java @@ -183,8 +183,8 @@ void setUp() throws Exception { contentClient = dev.openan.workflow.engine.examples.util.A2ATInitialization.create( () -> new net.openan.a2at.sdk.client.A2ATClient(java.nio.file.Path.of(sdkEnvPath))); - authorizationSender = new DefaultExtensionSender(authorizationTransport); - notificationSender = new DefaultExtensionSender(notificationTransport); + authorizationSender = DefaultExtensionSender.nonOwning(authorizationTransport); + notificationSender = DefaultExtensionSender.nonOwning(notificationTransport); } @AfterEach @@ -327,7 +327,6 @@ public void onEvent(String type, Map data) { .engineClient(client) .runtimeIntent("SPN跨城专线故障诊断与抢通：客户A上海-广州间SPN专线中断") .lang("zh") - .sslVerify(false) .eventCallback(cb) .execute() .join(); diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/HostQuickStartTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/HostQuickStartTest.java index b7cd2f5d..c1407d79 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/HostQuickStartTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/HostQuickStartTest.java @@ -33,7 +33,8 @@ void publishedHostExampleReachesLocalAggregation() throws Exception { var client = (WorkflowEngineClient) java.lang.reflect.Proxy.newProxyInstance(getClass().getClassLoader(), new Class[] {WorkflowEngineClient.class}, (proxy, method, args) -> { if (method.getName().equals("callbackTimeoutSeconds")) return 10L; - if (method.getName().equals("sendTask") && method.getParameterCount() == 3) { + if (method.getName().equals("sendTask") + && (method.getParameterCount() == 3 || method.getParameterCount() == 4)) { sent.incrementAndGet(); return CompletableFuture.completedFuture(SendMessageResult.builder().taskState("TASK_STATE_COMPLETED") .receivedMessages(List.of(new ReceivedMessage(MessageContent.text("diagnosis"), Map.of(), List.of()))) diff --git a/workflow-engine/pom.xml b/workflow-engine/pom.xml index 2c0bb4eb..d70d6abb 100644 --- a/workflow-engine/pom.xml +++ b/workflow-engine/pom.xml @@ -25,6 +25,33 @@ a2a-java-sdk-client ${a2a.java.sdk.version} + + + org.a2aproject.sdk + a2a-java-sdk-common + ${a2a.java.sdk.version} + + + org.a2aproject.sdk + a2a-java-sdk-spec + ${a2a.java.sdk.version} + + + org.a2aproject.sdk + a2a-java-sdk-client-transport-spi + ${a2a.java.sdk.version} + + + org.a2aproject.sdk + a2a-java-sdk-http-client + ${a2a.java.sdk.version} + + + org.a2aproject.sdk + a2a-java-sdk-jsonrpc-common + ${a2a.java.sdk.version} + org.a2aproject.sdk a2a-java-sdk-client-transport-rest @@ -45,9 +72,24 @@ io.grpc grpc-netty-shaded - 1.81.0 + ${grpc.version} runtime + + io.grpc + grpc-api + ${grpc.version} + + + com.google.protobuf + protobuf-java + ${protobuf.java.util.version} + + + com.google.protobuf + protobuf-java-util + ${protobuf.java.util.version} + net.openan.a2a-t.sdk @@ -61,6 +103,14 @@ jackson-databind ${jackson.version} + + com.fasterxml.jackson.core + jackson-core + + + com.fasterxml.jackson.core + jackson-annotations + diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java index a81bcfaa..60bc40f4 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java @@ -42,6 +42,7 @@ */ public interface A2AJavaClientRuntime { String CHANNEL_STATE_KEY = "dev.openan.workflow.engine.channel"; + String TRANSPORT_ACTIVITY_STATE_KEY = "dev.openan.workflow.engine.transportActivity"; String NOTIFICATION_CHANNEL = "notification"; /** diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java index 3b87d049..4630e55b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATransport.java @@ -35,6 +35,9 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -81,6 +84,7 @@ public class A2ATransport implements AutoCloseable { private final String contextId; private final ClientCallContextFactory clientCallContextFactory; private final ExecutorService asyncExecutor; + private final ScheduledExecutorService timeoutScheduler; private final long notificationAckTimeoutSeconds; private final long sendTimeoutSeconds; private final AtomicBoolean closed = new AtomicBoolean(); @@ -125,6 +129,13 @@ public A2ATransport( return t; }, new ThreadPoolExecutor.CallerRunsPolicy()); + this.timeoutScheduler = + Executors.newSingleThreadScheduledExecutor( + runnable -> { + Thread thread = new Thread(runnable, "transport-timeouts"); + thread.setDaemon(true); + return thread; + }); CredentialHttpTransport credentialHttpTransport = null; String credentialEncryptionKey = config.getCredentialEncryptionKey(); if (config.getCredentialsConfigPath() != null || config.getCredentialsConfig() != null) { @@ -578,6 +589,11 @@ public NotificationSubscription openNotificationStream( .put( A2AJavaClientRuntime.CHANNEL_STATE_KEY, A2AJavaClientRuntime.NOTIFICATION_CHANNEL); + callContext + .getState() + .put( + A2AJavaClientRuntime.TRANSPORT_ACTIVITY_STATE_KEY, + (Runnable) subscription::recordActivity); String endpoint = agentCard.supportedInterfaces().isEmpty() ? "?" @@ -653,7 +669,8 @@ public NotificationSubscription openNotificationStream( streamThread.setDaemon(true); streamThreadRef.set(streamThread); streamThread.start(); - CompletableFuture.runAsync( + ScheduledFuture acknowledgementTimeout = + timeoutScheduler.schedule( () -> { if (subscription.isActive() && !subscription.acknowledgement().isDone()) { log.warn( @@ -665,7 +682,12 @@ public NotificationSubscription openNotificationStream( subscription.close(); } }, - CompletableFuture.delayedExecutor(notificationAckTimeoutSeconds, TimeUnit.SECONDS)); + notificationAckTimeoutSeconds, + TimeUnit.SECONDS); + subscription + .acknowledgement() + .whenComplete((ignored, error) -> acknowledgementTimeout.cancel(false)); + subscription.completion().whenComplete((ignored, error) -> acknowledgementTimeout.cancel(false)); return subscription; } @@ -848,6 +870,7 @@ public void close() { } awaitNotificationShutdown(subscriptions); notificationSubscriptions.clear(); + timeoutScheduler.shutdownNow(); asyncExecutor.shutdown(); try { if (!asyncExecutor.awaitTermination(2, TimeUnit.SECONDS)) { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentAuthManager.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentAuthManager.java index f5de0275..5b169d1c 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentAuthManager.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentAuthManager.java @@ -31,7 +31,7 @@ * Loads agent credentials from config, creates per-agent CredentialService, and builds * auth/extension interceptors from AgentCard security schemes. * - *

Mirrors the Python SDK's {@code AgentAuthManager} + {@code AuthManager}. + *

Credential resolution is request-scoped; the manager does not own host token policy. */ class AgentAuthManager { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java index fd35d6af..efe072fc 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java @@ -325,6 +325,23 @@ public Iterable sendMessage( ClientCallContext callContext, Consumer eventSink, Consumer logSink) { + Object configuredActivity = + callContext == null + ? null + : callContext.getState().get(A2AJavaClientRuntime.TRANSPORT_ACTIVITY_STATE_KEY); + Runnable activityListener = + configuredActivity instanceof Runnable runnable ? runnable : null; + return TransportActivityMonitor.call( + activityListener, + () -> sendMessageObserved(agentCard, params, callContext, eventSink, logSink)); + } + + private Iterable sendMessageObserved( + AgentCard agentCard, + org.a2aproject.sdk.spec.MessageSendParams params, + ClientCallContext callContext, + Consumer eventSink, + Consumer logSink) { if (closed.get()) throw new IllegalStateException("A2A client runtime is closed"); String agentUrl = extractAgentUrl(agentCard); Client client = diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultExtensionSender.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultExtensionSender.java index a3ac05c7..2679b012 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultExtensionSender.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultExtensionSender.java @@ -25,14 +25,33 @@ import java.util.Objects; import java.util.UUID; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import org.a2aproject.sdk.spec.AgentCard; -/** Final-content sender on a caller-owned transport, independent of the workflow client. */ -public record DefaultExtensionSender(A2ATransport transport) - implements ExtensionSender, AutoCloseable { - public DefaultExtensionSender { - Objects.requireNonNull(transport, "transport"); +/** Final-content sender independent of the workflow client. */ +public final class DefaultExtensionSender implements ExtensionSender, AutoCloseable { + private final A2ATransport transport; + private final boolean closeTransportOnClose; + private final AtomicBoolean closed = new AtomicBoolean(); + + /** Creates a sender that owns and closes the supplied transport. */ + public DefaultExtensionSender(A2ATransport transport) { + this(transport, true); + } + + private DefaultExtensionSender(A2ATransport transport, boolean owning) { + this.transport = Objects.requireNonNull(transport, "transport"); + this.closeTransportOnClose = owning; + } + + /** Creates a sender facade whose transport remains caller-owned. */ + public static DefaultExtensionSender nonOwning(A2ATransport transport) { + return new DefaultExtensionSender(transport, false); + } + + public A2ATransport transport() { + return transport; } @Override @@ -79,6 +98,8 @@ private AgentCard requireExtension( @Override public void close() { - // Caller owns the independent transport and its subscriptions. + if (closed.compareAndSet(false, true) && closeTransportOnClose) { + transport.close(); + } } } 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 1a8fdfa4..a04b09a4 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 @@ -29,9 +29,13 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import net.openan.a2at.sdk.core.model.NegotiationContext; import net.openan.a2at.sdk.core.model.NegotiationPerformative; import org.a2aproject.sdk.client.ClientEvent; @@ -55,6 +59,13 @@ public class DefaultWorkflowEngineClient implements WorkflowEngineClient, AutoCl private final boolean closeTransportOnClose; private final AtomicBoolean closed = new AtomicBoolean(); private final Set invocations = java.util.concurrent.ConcurrentHashMap.newKeySet(); + private final ScheduledExecutorService timeoutScheduler = + Executors.newSingleThreadScheduledExecutor( + runnable -> { + Thread thread = new Thread(runnable, "workflow-client-timeouts"); + thread.setDaemon(true); + return thread; + }); private volatile EventCallback eventCallback = new EventCallback(); private volatile ControlPoint controlPoint; @@ -145,12 +156,21 @@ public CompletableFuture sendTask( .executionId(UUID.randomUUID().toString()) .taskId(UUID.randomUUID().toString()) .build(); - return dispatchTask(request, content, controlPoint); + return dispatchTask(request, content, controlPoint, eventCallback); } @Override public CompletableFuture sendTask( String agentName, MessageContent content, NegotiationStrategy negotiationStrategy) { + return sendTask(agentName, content, negotiationStrategy, eventCallback); + } + + @Override + public CompletableFuture sendTask( + String agentName, + MessageContent content, + NegotiationStrategy negotiationStrategy, + EventCallback invocationEvents) { if (negotiationStrategy == null) { return CompletableFuture.failedFuture(new NullPointerException("negotiationStrategy")); } @@ -162,11 +182,18 @@ public CompletableFuture sendTask( .build(); ControlPoint callbacks = ControlPoint.builder().onNegotiation(negotiationStrategy::resolve).build(); - return dispatchTask(request, content, callbacks); + return dispatchTask( + request, + content, + callbacks, + invocationEvents == null ? new EventCallback() : invocationEvents); } private CompletableFuture dispatchTask( - TaskRequest request, MessageContent content, ControlPoint callbacks) { + TaskRequest request, + MessageContent content, + ControlPoint callbacks, + EventCallback invocationEvents) { if (request == null) { return CompletableFuture.failedFuture(new NullPointerException("request")); } @@ -186,10 +213,10 @@ private CompletableFuture dispatchTask( } catch (RuntimeException error) { return CompletableFuture.failedFuture(error); } - Invocation invocation = new Invocation(request, content, callbacks); + Invocation invocation = new Invocation(request, content, callbacks, invocationEvents); invocations.add(invocation); - CompletableFuture.delayedExecutor(callbackTimeoutSeconds(), TimeUnit.SECONDS) - .execute( + ScheduledFuture timeout = + timeoutScheduler.schedule( () -> { if (!invocation.completion.isDone()) { completeExceptionallyAfterCleanup( @@ -200,9 +227,13 @@ private CompletableFuture dispatchTask( + callbackTimeoutSeconds() + " seconds")); } - }); + }, + callbackTimeoutSeconds(), + TimeUnit.SECONDS); invocation.completion.whenComplete( (result, error) -> { + timeout.cancel(false); + invocation.cancelPendingPoll(); invocations.remove(invocation); if (invocation.completion.isCancelled()) { cancelRemoteTask(card, invocation, null) @@ -214,6 +245,7 @@ private CompletableFuture dispatchTask( } if (error != null && invocation.exchanges > 0) emit( + invocation, EventType.NEGOTIATION_FAILED, Map.of( "agent", @@ -312,6 +344,7 @@ private CompletableFuture send( return CompletableFuture.failedFuture( new java.util.concurrent.CancellationException("Task interaction is no longer active")); emit( + invocation, EventType.AGENT_REQUEST, Map.of("agent", invocation.task.getAgentName(), "content", content)); Map trace = invocationTrace(invocation); @@ -324,7 +357,9 @@ private CompletableFuture send( content, invocation.contextId, taskId, - event -> forwardIntermediateEvent(event, invocation.task.getAgentName()), + event -> + forwardIntermediateEvent( + event, invocation.task.getAgentName(), invocation.events), invocation::active)); } @@ -371,7 +406,7 @@ private CompletableFuture advance( "Unsupported non-final remote task state: " + result.getTaskState())); } if (!"TASK_STATE_INPUT_REQUIRED".equals(result.getTaskState())) { - emitAgentResponse(invocation.task.getAgentName(), result); + emitAgentResponse(invocation, result); return CompletableFuture.completedFuture(result); } if (result.getTask() == null || result.getTask().id() == null) { @@ -405,6 +440,7 @@ private CompletableFuture advance( new NegotiationRequest( invocation.task, invocation.original, received, history, invocation.remaining()); emit( + invocation, EventType.NEGOTIATION_REQUEST, Map.of( "agent", @@ -437,6 +473,7 @@ private CompletableFuture advance( MessageContent content = ((NegotiationReply.Send) reply).content(); NegotiationContext ending = validateReply(context, content); emit( + invocation, EventType.NEGOTIATION_RESOLVED, Map.of( "agent", @@ -453,7 +490,7 @@ private CompletableFuture advance( next.setFailureCode("negotiation.aborted"); next.setFailureMessage( "Business sent Abort; diagnosis was not completed successfully"); - emitAgentResponse(invocation.task.getAgentName(), next); + emitAgentResponse(invocation, next); return CompletableFuture.completedFuture(next); } return advance(card, invocation, next); @@ -466,13 +503,7 @@ private CompletableFuture observeTask(AgentCard card, Invocat if (invocation.remoteTaskId == null) return CompletableFuture.failedFuture( new IllegalArgumentException("Non-terminal response has no remote task identity")); - return CompletableFuture.supplyAsync( - () -> { - if (!invocation.active()) - throw new java.util.concurrent.CancellationException("Task wait ended"); - return Boolean.TRUE; - }, - CompletableFuture.delayedExecutor(250, TimeUnit.MILLISECONDS)) + return delayBeforeTaskPoll(invocation) .thenCompose( ignored -> { if (!invocation.active()) @@ -486,23 +517,41 @@ private CompletableFuture observeTask(AgentCard card, Invocat }); } - private void emit(String type, Map data) { + private CompletableFuture delayBeforeTaskPoll(Invocation invocation) { + PendingPoll pending = new PendingPoll(); + PendingPoll previous = invocation.pendingPoll.getAndSet(pending); + if (previous != null) previous.cancel(); + pending.scheduled = + timeoutScheduler.schedule( + () -> { + if (invocation.pendingPoll.compareAndSet(pending, null)) { + pending.completion.complete(null); + } + }, + 250, + TimeUnit.MILLISECONDS); + if (pending.completion.isDone()) pending.scheduled.cancel(false); + return pending.completion; + } + + private void emit(Invocation invocation, String type, Map data) { try { - eventCallback.onEvent(type, data); + invocation.events.onEvent(type, data); } catch (RuntimeException error) { log.warn("Event callback failed for {}", type, error); } } - private void emitAgentResponse(String agentName, SendMessageResult result) { + private void emitAgentResponse(Invocation invocation, SendMessageResult result) { Map data = new HashMap<>(); - data.put("agent", agentName); + data.put("agent", invocation.task.getAgentName()); data.put("response", result.getText()); data.put("receivedMessages", result.getReceivedMessages()); - emit(EventType.AGENT_RESPONSE, data); + emit(invocation, EventType.AGENT_RESPONSE, data); } - private void forwardIntermediateEvent(ClientEvent event, String agentName) { + private void forwardIntermediateEvent( + ClientEvent event, String agentName, EventCallback invocationEvents) { Map data = ClientEventMapper.toMap(event, agentName); if (event instanceof TaskUpdateEvent tue) { if (tue.getUpdateEvent() instanceof TaskStatusUpdateEvent sue) { @@ -512,7 +561,7 @@ private void forwardIntermediateEvent(ClientEvent event, String agentName) { agentName, state, sue.isFinal()); - emit(EventType.AGENT_STATUS_UPDATE, data); + emit(invocationEvents, EventType.AGENT_STATUS_UPDATE, data); } else if (tue.getUpdateEvent() instanceof org.a2aproject.sdk.spec.TaskArtifactUpdateEvent ae) { log.info( @@ -520,12 +569,20 @@ private void forwardIntermediateEvent(ClientEvent event, String agentName) { agentName, ae.artifact().name(), ae.artifact().artifactId()); - emit(EventType.AGENT_ARTIFACT_UPDATE, data); + emit(invocationEvents, EventType.AGENT_ARTIFACT_UPDATE, data); } } else if (event instanceof MessageEvent) { String text = (String) data.getOrDefault("text", ""); log.info("[EngineClient] Agent {} message event: {} chars", agentName, text.length()); - emit(EventType.AGENT_MESSAGE_EVENT, data); + emit(invocationEvents, EventType.AGENT_MESSAGE_EVENT, data); + } + } + + private void emit(EventCallback callback, String type, Map data) { + try { + callback.onEvent(type, data); + } catch (RuntimeException error) { + log.warn("Event callback failed for {}", type, error); } } @@ -594,7 +651,7 @@ public CompletableFuture subscribeToTask( agentName, taskId, event -> { - forwardIntermediateEvent(event, agentName); + forwardIntermediateEvent(event, agentName, this.eventCallback); if (eventCallback != null) { eventCallback.accept(ClientEventMapper.toMap(event, agentName)); } @@ -605,6 +662,7 @@ public CompletableFuture subscribeToTask( public void close() { if (!closed.compareAndSet(false, true)) return; invocations.forEach(invocation -> invocation.completion.cancel(true)); + timeoutScheduler.shutdownNow(); if (closeTransportOnClose) transport.close(); } @@ -612,6 +670,7 @@ private final class Invocation { final TaskRequest task; final MessageContent original; final ControlPoint callbacks; + final EventCallback events; final String contextId = UUID.randomUUID().toString(); final String attempt = UUID.randomUUID().toString(); final Map parentTrace = Map.copyOf(WireLog.context()); @@ -621,15 +680,21 @@ private final class Invocation { final Set answered = new HashSet<>(); final Map contexts = new HashMap<>(); final AtomicBoolean finishing = new AtomicBoolean(); + final AtomicReference pendingPoll = new AtomicReference<>(); volatile String remoteTaskId; volatile boolean remoteTaskTerminal; CompletableFuture cleanup; int exchanges; - Invocation(TaskRequest task, MessageContent content, ControlPoint callbacks) { + Invocation( + TaskRequest task, + MessageContent content, + ControlPoint callbacks, + EventCallback events) { this.task = task; this.original = content; this.callbacks = callbacks; + this.events = events; } Duration remaining() { @@ -642,5 +707,22 @@ boolean active() { && !completion.isDone() && !remaining().isZero(); } + + void cancelPendingPoll() { + PendingPoll pending = pendingPoll.getAndSet(null); + if (pending != null) pending.cancel(); + } + } + + private static final class PendingPoll { + final CompletableFuture completion = new CompletableFuture<>(); + volatile ScheduledFuture scheduled; + + void cancel() { + ScheduledFuture current = scheduled; + if (current != null) current.cancel(false); + completion.completeExceptionally( + new java.util.concurrent.CancellationException("Task wait ended")); + } } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/NotificationSubscription.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/NotificationSubscription.java index 0ab7452e..ee8e908d 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/NotificationSubscription.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/NotificationSubscription.java @@ -45,7 +45,8 @@ public final class NotificationSubscription implements AutoCloseable { private final Runnable closeAction; private final AtomicBoolean closed = new AtomicBoolean(); private final AtomicLong eventCount = new AtomicLong(); - private final AtomicReference lastEventAt = new AtomicReference<>(); + private final AtomicReference lastActivityAt = new AtomicReference<>(); + private final AtomicReference lastBusinessEventAt = new AtomicReference<>(); private volatile Throwable streamFailure; NotificationSubscription(String agentName, String contextId, Runnable closeAction) { @@ -74,9 +75,18 @@ public boolean isActive() { return !closed.get() && !completion.isDone(); } - /** Current local liveness snapshot; every protocol event, including heartbeats, refreshes it. */ + /** + * Current local liveness snapshot. For compatibility, {@link Heartbeat#lastEventAt()} represents + * the last transport activity, including an SSE comment heartbeat; {@code eventCount} counts only + * decoded A2A business events. + */ public Heartbeat heartbeat() { - return new Heartbeat(openedAt, lastEventAt.get(), eventCount.get(), isActive()); + return new Heartbeat(openedAt, lastActivityAt.get(), eventCount.get(), isActive()); + } + + /** Last decoded A2A business event, excluding transport-only SSE comments. */ + public Instant lastBusinessEventAt() { + return lastBusinessEventAt.get(); } public boolean isHealthy(Duration maximumIdle) { @@ -84,7 +94,7 @@ public boolean isHealthy(Duration maximumIdle) { if (maximumIdle.isNegative()) { throw new IllegalArgumentException("maximumIdle must not be negative"); } - Instant activity = lastEventAt.get(); + Instant activity = lastActivityAt.get(); if (!isActive() || activity == null) { return false; } @@ -93,7 +103,13 @@ public boolean isHealthy(Duration maximumIdle) { void recordEvent() { eventCount.incrementAndGet(); - lastEventAt.set(Instant.now()); + Instant now = Instant.now(); + lastBusinessEventAt.set(now); + lastActivityAt.set(now); + } + + void recordActivity() { + lastActivityAt.set(Instant.now()); } void acknowledge(SendMessageResult result) { @@ -137,6 +153,9 @@ public void close() { } } - /** Liveness snapshot of one subscription: open time, last event, event count, active flag. */ + /** + * Liveness snapshot of one subscription: open time, last transport activity (legacy accessor + * name), decoded A2A event count, and active flag. + */ public record Heartbeat(Instant openedAt, Instant lastEventAt, long eventCount, boolean active) {} } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java index ab78645e..d6d07f4b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ObservedHttpClient.java @@ -144,6 +144,7 @@ private final class Exchange { final String id = UUID.randomUUID().toString(); final HttpRequest request; final Map context = WireLog.context(); + final Runnable activityListener = TransportActivityMonitor.capture(); final java.util.concurrent.atomic.AtomicBoolean failureLogged = new java.util.concurrent.atomic.AtomicBoolean(); final java.util.concurrent.atomic.AtomicBoolean cancelledBySubscriber = @@ -329,6 +330,7 @@ public void cancel() { @Override public void onNext(List items) { + TransportActivityMonitor.notifyActivity(activityListener); items.forEach(body::accept); target.onNext(items); } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportActivityMonitor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportActivityMonitor.java new file mode 100644 index 00000000..848cb5ee --- /dev/null +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportActivityMonitor.java @@ -0,0 +1,57 @@ +/* + * 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.client; + +import java.util.function.Supplier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Captures a per-request transport activity callback for asynchronous HTTP body observation. */ +final class TransportActivityMonitor { + private static final Logger log = LoggerFactory.getLogger(TransportActivityMonitor.class); + private static final ThreadLocal CURRENT = new ThreadLocal<>(); + + private TransportActivityMonitor() {} + + static T call(Runnable listener, Supplier action) { + Runnable previous = CURRENT.get(); + if (listener == null) CURRENT.remove(); + else CURRENT.set(listener); + try { + return action.get(); + } finally { + if (previous == null) CURRENT.remove(); + else CURRENT.set(previous); + } + } + + static Runnable capture() { + return CURRENT.get(); + } + + static void notifyActivity(Runnable listener) { + if (listener == null) return; + try { + listener.run(); + } catch (RuntimeException error) { + log.warn("Transport activity callback failed", error); + } + } +} diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java index 45bca8a7..ef6d97eb 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java @@ -45,6 +45,19 @@ public interface WorkflowEngineClient { CompletableFuture sendTask( String agentName, MessageContent content, NegotiationStrategy negotiationStrategy); + /** + * Sends final task content with callbacks scoped to this invocation. Implementations that emit + * protocol lifecycle events should override this method so concurrent workflow executions do not + * share mutable callback state. + */ + default CompletableFuture sendTask( + String agentName, + MessageContent content, + NegotiationStrategy negotiationStrategy, + EventCallback eventCallback) { + return sendTask(agentName, content, negotiationStrategy); + } + /** * Maximum wait for one task interaction, from initial content preparation through remote task * completion and any negotiation exchanges. The same budget applies independently to a route 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 4d50da5f..f8b83721 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 @@ -29,8 +29,8 @@ /** * Configuration for DefaultWorkflowEngineClient. * - *

Mirrors the Python SDK's {@code WorkflowEngineClient.__init__} parameters. Use the builder to - * create an instance. + *

Use the builder to create immutable transport, authentication, timeout, and + * protocol-selection settings for one engine client. */ @Getter public class WorkflowEngineClientConfig { @@ -235,6 +235,10 @@ public WorkflowEngineClientConfig build() { if (maxNegotiationExchanges <= 0) { throw new IllegalArgumentException("maxNegotiationExchanges must be positive"); } + if (credentialsConfigPath != null && credentialsConfig != null) { + throw new IllegalArgumentException( + "credentialsConfigPath and credentialsConfig are mutually exclusive"); + } return new WorkflowEngineClientConfig(this); } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java index c0acf141..1c398739 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java @@ -100,8 +100,6 @@ public WorkflowExecutor( this.eventCallback = eventCallback != null ? eventCallback : new EventCallback(); this.contextBuilder = new ContextBuilder(this.workflow, runtimeIntent); this.lang = lang != null ? lang : "zh"; - this.engineClient.setControlPoint(this.controlPoint); - this.engineClient.setEventCallback(this.eventCallback); log.info( "[Executor] Workflow: {}, steps={}, intent={}, lang={}", workflow.getName(), @@ -403,7 +401,8 @@ private CompletableFuture dispatchTask(WorkflowStep step, TaskReques negotiation.originalSubmission(), negotiation.received(), negotiation.previousExchanges(), - negotiation.remainingWait())))); + negotiation.remainingWait())), + eventCallback)); result.whenComplete( (value, failure) -> { if (!sent.isDone()) sent.cancel(true); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java index 6c33ce7f..1aba6d04 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java @@ -30,6 +30,7 @@ import java.net.URI; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.List; import java.util.Map; import javax.net.ssl.HttpsURLConnection; @@ -51,9 +52,22 @@ public class LoadPsop { private static final Logger log = LoggerFactory.getLogger(LoadPsop.class); private static final ObjectMapper mapper = new ObjectMapper(); + private static final Timeouts DEFAULT_TIMEOUTS = + new Timeouts(Duration.ofSeconds(30), Duration.ofSeconds(30)); public static Workflow load(String baseUrl, String psopId, String accessToken, boolean sslVerify) throws Exception { + return load(baseUrl, psopId, accessToken, sslVerify, DEFAULT_TIMEOUTS); + } + + /** Loads one workflow with explicit connection and response-read deadlines. */ + public static Workflow load( + String baseUrl, + String psopId, + String accessToken, + boolean sslVerify, + Timeouts timeouts) + throws Exception { StringBuilder urlBuilder = new StringBuilder(baseUrl).append("/api/v1/orchestrate/psop/").append(psopId); if (accessToken != null && !accessToken.isEmpty()) { @@ -63,7 +77,7 @@ public static Workflow load(String baseUrl, String psopId, String accessToken, b } String url = urlBuilder.toString(); log.info("[Registry] Loading PSOP from {} (ssl_verify={})", anonymousUrl(url, accessToken), sslVerify); - HttpResult resp = execute("GET", url, null, sslVerify); + HttpResult resp = execute("GET", url, null, sslVerify, timeouts); if (resp.statusCode() != 200) { throw requestFailure(resp); } @@ -83,6 +97,18 @@ public static Workflow load(String baseUrl, String psopId) throws Exception { public static List search( String baseUrl, String intent, int topN, String accessToken, boolean sslVerify) throws Exception { + return search(baseUrl, intent, topN, accessToken, sslVerify, DEFAULT_TIMEOUTS); + } + + /** Searches workflows with explicit connection and response-read deadlines. */ + public static List search( + String baseUrl, + String intent, + int topN, + String accessToken, + boolean sslVerify, + Timeouts timeouts) + throws Exception { StringBuilder urlBuilder = new StringBuilder(baseUrl).append("/api/v1/orchestrate/search"); if (accessToken != null && !accessToken.isEmpty()) { urlBuilder @@ -92,7 +118,7 @@ public static List search( String url = urlBuilder.toString(); log.info("[Registry] Searching PSOP at {} (intent={}, top_n={})", anonymousUrl(url, accessToken), intent, topN); String jsonBody = mapper.writeValueAsString(Map.of("intent", intent, "top_n", topN)); - HttpResult resp = execute("POST", url, jsonBody, sslVerify); + HttpResult resp = execute("POST", url, jsonBody, sslVerify, timeouts); if (resp.statusCode() != 200) { throw requestFailure(resp); } @@ -137,8 +163,14 @@ private static RuntimeException requestFailure(HttpResult response) { + (detail.isBlank() ? "" : ": " + detail)); } - private static HttpResult execute(String method, String url, String jsonBody, boolean sslVerify) + private static HttpResult execute( + String method, + String url, + String jsonBody, + boolean sslVerify, + Timeouts timeouts) throws Exception { + java.util.Objects.requireNonNull(timeouts, "timeouts"); HttpURLConnection connection = (HttpURLConnection) URI.create(url).toURL().openConnection(); if (!sslVerify && connection instanceof HttpsURLConnection https) { // Development opt-out is connection-local: never replace JVM-wide TLS defaults. @@ -152,8 +184,8 @@ private static HttpResult execute(String method, String url, String jsonBody, bo } try { connection.setInstanceFollowRedirects(false); - connection.setConnectTimeout(30_000); - connection.setReadTimeout(30_000); + connection.setConnectTimeout(timeouts.connectTimeoutMillis()); + connection.setReadTimeout(timeouts.readTimeoutMillis()); connection.setRequestMethod(method); if (jsonBody != null) { connection.setDoOutput(true); @@ -179,5 +211,39 @@ public static List search(String baseUrl, String intent) t return search(baseUrl, intent, 5, null, true); } + /** Connection and response-read deadlines for orchestration-center calls. */ + public record Timeouts(Duration connectTimeout, Duration readTimeout) { + public Timeouts { + requirePositiveAndSupported(connectTimeout, "connectTimeout"); + requirePositiveAndSupported(readTimeout, "readTimeout"); + } + + int connectTimeoutMillis() { + return Math.toIntExact(connectTimeout.toMillis()); + } + + int readTimeoutMillis() { + return Math.toIntExact(readTimeout.toMillis()); + } + + private static void requirePositiveAndSupported(Duration value, String name) { + if (value == null || value.isZero() || value.isNegative()) { + throw new IllegalArgumentException(name + " must be positive"); + } + long milliseconds; + try { + milliseconds = value.toMillis(); + } catch (ArithmeticException error) { + throw new IllegalArgumentException(name + " exceeds HttpURLConnection limit", error); + } + if (milliseconds < 1) { + throw new IllegalArgumentException(name + " must be at least 1 millisecond"); + } + if (milliseconds > Integer.MAX_VALUE) { + throw new IllegalArgumentException(name + " exceeds HttpURLConnection limit"); + } + } + } + private record HttpResult(int statusCode, String body) {} } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java index 032f3ab2..31a95812 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java @@ -412,6 +412,7 @@ public static final class Builder { private String caCertsPath; private A2AJavaClientRuntime a2aClientRuntime; private EventCallback eventCallback; + private boolean clientConstructionConfigured; private BiFunction>, CompletableFuture> onFinish; private Function, Object> onEvent; @@ -423,6 +424,7 @@ public Builder psop(Workflow psop) { public Builder agentCards(List v) { this.agentCards = v; + this.clientConstructionConfigured = true; return this; } @@ -448,21 +450,25 @@ public Builder lang(String v) { public Builder credentialsConfigPath(String v) { this.credentialsConfigPath = v; + this.clientConstructionConfigured = true; return this; } public Builder sslVerify(boolean v) { this.sslVerify = v; + this.clientConstructionConfigured = true; return this; } public Builder caCertsPath(String v) { this.caCertsPath = v; + this.clientConstructionConfigured = true; return this; } public Builder a2aClientRuntime(A2AJavaClientRuntime v) { this.a2aClientRuntime = v; + this.clientConstructionConfigured = true; return this; } @@ -496,6 +502,10 @@ public Builder onEvent(Function, Object> v) { public CompletableFuture execute() { if (psop == null) throw new IllegalArgumentException("psop is required"); if (controlPoint == null) throw new IllegalArgumentException("controlPoint is required"); + if (engineClient != null && clientConstructionConfigured) { + throw new IllegalArgumentException( + "engineClient cannot be combined with agentCards, a2aClientRuntime, TLS, or credentials settings"); + } return ExecutePsop.execute( psop, agentCards, 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 777c0ed7..59b4a61c 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 @@ -678,4 +678,66 @@ void closingNonOwningFacadePreventsNewWorkButDoesNotCloseSharedTransport() throw .getTaskState()); } } + + @Test + void concurrentTasksKeepInvocationEventCallbacksIsolated() throws Exception { + CountDownLatch bothStarted = new CountDownLatch(2); + A2AJavaClientRuntime concurrentRuntime = + runtime( + params -> { + bothStarted.countDown(); + try { + assertTrue(bothStarted.await(2, TimeUnit.SECONDS)); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + throw new CompletionException(error); + } + return List.of( + response(params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); + }); + try (var transport = + new A2ATransport( + List.of(card()), concurrentRuntime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + List firstEvents = new CopyOnWriteArrayList<>(); + List secondEvents = new CopyOnWriteArrayList<>(); + List firstRequests = new CopyOnWriteArrayList<>(); + List secondRequests = new CopyOnWriteArrayList<>(); + + CompletableFuture first = + client.sendTask( + "test", + MessageContent.text("first"), + request -> CompletableFuture.failedFuture(new AssertionError("not negotiated")), + new dev.openan.workflow.engine.control.EventCallback() { + @Override + public void onEvent(String type, Map data) { + firstEvents.add(type); + if ("agent_request".equals(type)) { + firstRequests.add((MessageContent) data.get("content")); + } + } + }); + CompletableFuture second = + client.sendTask( + "test", + MessageContent.text("second"), + request -> CompletableFuture.failedFuture(new AssertionError("not negotiated")), + new dev.openan.workflow.engine.control.EventCallback() { + @Override + public void onEvent(String type, Map data) { + secondEvents.add(type); + if ("agent_request".equals(type)) { + secondRequests.add((MessageContent) data.get("content")); + } + } + }); + + CompletableFuture.allOf(first, second).join(); + assertEquals(List.of("agent_request", "agent_response"), firstEvents); + assertEquals(List.of("agent_request", "agent_response"), secondEvents); + assertEquals(List.of(MessageContent.text("first")), firstRequests); + assertEquals(List.of(MessageContent.text("second")), secondRequests); + } + } } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ExtensionSenderTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ExtensionSenderTest.java index 96d4ffaf..8a6fd9c4 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ExtensionSenderTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ExtensionSenderTest.java @@ -132,4 +132,56 @@ public void close() {} release.countDown(); } } + + @Test + void closeHonorsExplicitTransportOwnership() throws Exception { + AtomicInteger owningCloses = new AtomicInteger(); + A2AJavaClientRuntime owningRuntime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + return List.of(); + } + + public void close() { + owningCloses.incrementAndGet(); + } + }; + A2ATransport ownedTransport = + new A2ATransport( + List.of(card()), owningRuntime, WorkflowEngineClientConfig.builder().build()); + DefaultExtensionSender owning = new DefaultExtensionSender(ownedTransport); + owning.close(); + owning.close(); + assertEquals(1, owningCloses.get()); + + AtomicInteger sharedCloses = new AtomicInteger(); + A2AJavaClientRuntime sharedRuntime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + return List.of(); + } + + public void close() { + sharedCloses.incrementAndGet(); + } + }; + A2ATransport sharedTransport = + new A2ATransport( + List.of(card()), sharedRuntime, WorkflowEngineClientConfig.builder().build()); + DefaultExtensionSender nonOwning = DefaultExtensionSender.nonOwning(sharedTransport); + nonOwning.close(); + assertEquals(0, sharedCloses.get()); + sharedTransport.close(); + assertEquals(1, sharedCloses.get()); + } } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/NotificationSubscriptionTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/NotificationSubscriptionTest.java index ff2f512f..a22ea663 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/NotificationSubscriptionTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/NotificationSubscriptionTest.java @@ -40,11 +40,17 @@ void eventsRefreshHeartbeatAndExplicitCloseIsIdempotent() { assertEquals(0, subscription.heartbeat().eventCount()); assertFalse(subscription.isHealthy(Duration.ofSeconds(1))); + subscription.recordActivity(); + assertEquals(0, subscription.heartbeat().eventCount()); + assertTrue(subscription.isHealthy(Duration.ofSeconds(1))); + assertEquals(null, subscription.lastBusinessEventAt()); + subscription.recordEvent(); subscription.acknowledge( SendMessageResult.builder().text("ack").taskState("TASK_STATE_WORKING").build()); assertEquals(1, subscription.heartbeat().eventCount()); + assertTrue(subscription.lastBusinessEventAt() != null); assertTrue(subscription.isHealthy(Duration.ofSeconds(1))); subscription.close(); subscription.close(); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java index 5991209d..9e734f0e 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ObservedHttpClientTest.java @@ -30,6 +30,7 @@ import java.time.Duration; import java.util.*; import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -128,6 +129,45 @@ void streamingFramesKeepUtf8ControlLinesAndArriveBeforeCompletion() { assertEquals(2, frames.size()); } + @Test + void rawSseCommentRefreshesTransportActivityWithoutNeedingADataEvent() throws Exception { + AtomicInteger activities = new AtomicInteger(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/events", + exchange -> { + byte[] bytes = ": heartbeat\n\n".getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + try { + HttpRequest request = + HttpRequest.newBuilder( + URI.create("http://127.0.0.1:" + server.getAddress().getPort() + "/events")) + .GET() + .build(); + HttpResponse response = + TransportActivityMonitor.call( + activities::incrementAndGet, + () -> { + try { + return new ObservedHttpClient(HttpClient.newHttpClient()) + .send(request, HttpResponse.BodyHandlers.ofString()); + } catch (IOException | InterruptedException error) { + throw new CompletionException(error); + } + }); + + assertEquals(": heartbeat\n\n", response.body()); + assertTrue(activities.get() > 0); + } finally { + server.stop(0); + } + } + @Test void oversizedFramesAreExplicitlyDroppedWithoutLeakingPartialSecrets() { System.setProperty("WORKFLOW_ENGINE_PROTOCOL_MAX_BODY_CHARS", "256"); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java index 7b873b73..5caea7ae 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java @@ -25,6 +25,23 @@ import org.junit.jupiter.api.Test; class WorkflowEngineClientConfigTest { + + @Test + void rejectsFileAndInlineCredentialsTogether() { + Map>> inline = + Map.of("agent", Map.of("scheme", Map.of("token", "value"))); + + IllegalArgumentException error = + assertThrows( + IllegalArgumentException.class, + () -> + WorkflowEngineClientConfig.builder() + .credentialsConfigPath("credentials.json") + .credentialsConfig(inline) + .build()); + + assertTrue(error.getMessage().contains("mutually exclusive")); + } @Test void notificationAcknowledgementDefaultsToFiveMinutesAndRemainsConfigurable() { assertEquals( diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java index 5aa3916b..9fe86144 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java @@ -20,6 +20,7 @@ package dev.openan.workflow.engine.core; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -107,7 +108,7 @@ public CompletableFuture onRoute(RouteRequest routeRequest) { } @Test - void constructorFailsFastWhenClientCallbackWiringFails() { + void constructorDoesNotMutateClientCallbacks() { Workflow workflow = Workflow.builder() .name("wiring") @@ -121,13 +122,17 @@ void constructorFailsFastWhenClientCallbackWiringFails() { public void setControlPoint(ControlPoint controlPoint) { throw new IllegalStateException("wiring failed"); } + + @Override + public void setEventCallback(EventCallback callback) { + throw new IllegalStateException("wiring failed"); + } }; - IllegalStateException error = - assertThrows( - IllegalStateException.class, + WorkflowExecutor executor = + assertDoesNotThrow( () -> new WorkflowExecutor(workflow, autoCp(), client, recordingCallback(), "", "zh")); - assertEquals("wiring failed", error.getMessage()); + assertTrue(executor.run().join().isSuccess()); } @Test diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java index 2f443f55..a8de6e4a 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java @@ -22,6 +22,7 @@ import static org.junit.jupiter.api.Assertions.*; import dev.openan.workflow.engine.model.WorkflowSearchResult; +import java.time.Duration; import java.util.List; import org.junit.jupiter.api.Test; @@ -31,6 +32,55 @@ */ class LoadPsopSearchTest { + @Test + void rejectsInvalidCustomTimeouts() { + assertThrows( + IllegalArgumentException.class, + () -> new LoadPsop.Timeouts(Duration.ZERO, Duration.ofSeconds(1))); + assertThrows( + IllegalArgumentException.class, + () -> new LoadPsop.Timeouts(Duration.ofSeconds(1), Duration.ofMillis(-1))); + assertThrows( + IllegalArgumentException.class, + () -> new LoadPsop.Timeouts(Duration.ofNanos(1), Duration.ofSeconds(1))); + } + + @Test + void customReadTimeoutIsApplied() throws Exception { + com.sun.net.httpserver.HttpServer server = + com.sun.net.httpserver.HttpServer.create(new java.net.InetSocketAddress(0), 0); + server.createContext( + "/api/v1/orchestrate/search", + exchange -> { + try { + Thread.sleep(500); + byte[] response = "{\"data\":[]}".getBytes(java.nio.charset.StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + } finally { + exchange.close(); + } + }); + server.start(); + try { + String baseUrl = "http://127.0.0.1:" + server.getAddress().getPort(); + assertThrows( + java.net.SocketTimeoutException.class, + () -> + LoadPsop.search( + baseUrl, + "timeout", + 1, + null, + true, + new LoadPsop.Timeouts(Duration.ofSeconds(1), Duration.ofMillis(50)))); + } finally { + server.stop(0); + } + } + @Test void tokenLogsRevealPresenceButNoTokenCharacters() { assertEquals("https://example.test/search?access_token=", diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java index 9a017cc9..2cc31202 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java @@ -356,12 +356,10 @@ void builderProducesSameResultAsStaticExecute() { ExecutionResult r2 = ExecutePsop.builder() .psop(linearWorkflow()) - .agentCards(List.of()) .controlPoint(autoCp()) .engineClient(stub2) .runtimeIntent("intent") .lang("zh") - .sslVerify(false) .execute() .join(); assertEquals(r1.isSuccess(), r2.isSuccess()); @@ -405,6 +403,22 @@ public void close() { assertEquals(0, closeCalls.get()); } + @Test + void builderRejectsInjectedClientCombinedWithConstructionSettings() { + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A", "B"); + IllegalArgumentException error = + assertThrows( + IllegalArgumentException.class, + () -> + ExecutePsop.builder() + .psop(linearWorkflow()) + .controlPoint(autoCp()) + .engineClient(client) + .sslVerify(false) + .execute()); + assertTrue(error.getMessage().contains("engineClient cannot be combined")); + } + @Test void errorLifecycleOnTaskFailure() { StubWorkflowEngineClient stub = new StubWorkflowEngineClient("A", "B"); From bb85f62466a761376e9734a080140ec50cf8812c Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Sun, 13 Sep 2026 17:06:37 +0800 Subject: [PATCH 2/2] docs: clarify lifecycle and configuration contracts Signed-off-by: Zhoujie628 --- docs/en/API_REFERENCE.md | 28 +++++++++++++++++++++++++--- docs/en/BUSINESS_CALLBACKS.md | 2 +- docs/en/INTEGRATION_GUIDE.md | 10 ++++++++++ docs/zh/API_REFERENCE.md | 23 ++++++++++++++++++++--- docs/zh/BUSINESS_CALLBACKS.md | 2 +- docs/zh/INTEGRATION_GUIDE.md | 8 ++++++++ 6 files changed, 65 insertions(+), 8 deletions(-) diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index a601c4f3..451dd3f3 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -27,7 +27,7 @@ Entry point for executing a PSOP workflow. Uses the Builder pattern. | `psop(Workflow)` | required | - | PSOP workflow definition | | `agentCards(List)` | optional | `List.of()` | Cards for dispatched agents; required for remote steps unless a configured `engineClient` supplies transport | | `controlPoint(ControlPoint)` | required | - | User decision implementation | -| `engineClient(WorkflowEngineClient)` | optional | null | Pre-configured client (null = auto-create) | +| `engineClient(WorkflowEngineClient)` | optional | null | Pre-configured client (null = auto-create); cannot be combined with AgentCard, runtime, TLS, or credentials construction settings | | `runtimeIntent(String)` | optional | `""` | Natural-language intent for context assembly | | `lang(String)` | optional | `"zh"` | Language hint (`"zh"` or `"en"`) | | `credentialsConfigPath(String)` | optional | null | Path to credentials JSON file | @@ -61,6 +61,8 @@ ExecutionResult result = ExecutePsop.builder() CompletableFuture sendTask(String agentName, MessageContent content); CompletableFuture sendTask(String agentName, MessageContent content, NegotiationStrategy negotiationStrategy); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy, EventCallback invocationEvents); CompletableFuture getTask(String agentName, String taskId); CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); @@ -75,6 +77,9 @@ void close(); AgentCard, authentication and custom runtime as message dispatch. A list result is authorization-scoped: it contains only tasks visible to the authenticated identity. Callers must paginate with `nextPageToken`; cancellation is not implied by closing a local client or notification stream. +The Future returned by `subscribeToTask` represents the subscription result, not an independently +closeable SSE handle. Cancelling it does not guarantee transport termination; close the owning Client/Runtime +to release a non-terminal subscription. `sendTask` executes final host-provided content outside a DAG. Every call has a fresh context and uses the same send, wait, task-query and Negotiation-T loop as workflow execution. The overload takes @@ -118,6 +123,13 @@ Authorization and notification accept host-generated final content. Use separate their outcomes do not gate the workflow. openNotification registers a handle before I/O, and passes it plus ReceivedMessage directly to the listener. acknowledgement() is the real ACK; timeout fails. close() requests closure; completion() observes actual stream termination. +For compatibility, `heartbeat().lastEventAt()` keeps its old name but represents the last transport activity. +A standard SSE `: heartbeat` refreshes it without increasing `eventCount`, which counts decoded A2A business events only. +`lastBusinessEventAt()` returns the most recent business-event time. + +`new DefaultExtensionSender(transport)` owns and closes the transport. Use +`DefaultExtensionSender.nonOwning(transport)` only when another component owns the transport, and then close subscriptions +and the transport explicitly. ### WorkflowEngineClientConfig @@ -142,6 +154,9 @@ Builder-based configuration for the workflow engine client. | `credentialsConfig` | `Map` | null | Inline credentials config; must match AgentCard security requirements | | `maxNegotiationExchanges` | `int` | `3` | Local interaction budget, independent of SDK maxRounds | +`credentialsConfigPath` and `credentialsConfig` are mutually exclusive built-in credential sources; +configuring both fails when the configuration is built. + ```java WorkflowEngineClientConfig config = WorkflowEngineClientConfig.builder() .sslVerify(true) @@ -279,7 +294,7 @@ runtime, diagnostic adapter, or explicit lifecycle requires them. | `A2ATExtension` | Canonical extension names and URIs | | `A2ATransport` | Low-level transport, authentication, response assembly, and subscription lifecycle | | `DefaultWorkflowEngineClient` | Default `WorkflowEngineClient` implementation | -| `DefaultExtensionSender` | Default `ExtensionSender` implementation backed by an `A2ATransport` | +| `DefaultExtensionSender` | Default `ExtensionSender`; its public constructor owns the transport, while `nonOwning` does not | | `DefaultA2AJavaClientRuntime` | Default A2A Java SDK runtime for HTTP/JSON-RPC/gRPC | | `CredentialCrypto` | AES-GCM credential encryption utility and command-line entry point | | `EnvFileLoader` | Explicit `.env` parser for host-owned configuration | @@ -365,6 +380,8 @@ Load and search PSOP workflows from the orchestration center. ```java static Workflow load(String baseUrl, String psopId, String accessToken, boolean sslVerify) +static Workflow load(String baseUrl, String psopId, + String accessToken, boolean sslVerify, LoadPsop.Timeouts timeouts) static Workflow load(String baseUrl, String psopId) ``` @@ -377,6 +394,9 @@ GET `/api/v1/orchestrate/psop/{psop_id}`. Returns the full workflow with steps, static List search( String baseUrl, String intent, int topN, String accessToken, boolean sslVerify) +static List search( + String baseUrl, String intent, int topN, + String accessToken, boolean sslVerify, LoadPsop.Timeouts timeouts) static List search( String baseUrl, String intent) @@ -389,6 +409,8 @@ Explicit `false` skips certificate-chain and hostname verification on that orche allowing development without a local CA file. The server still needs a TLS certificate; this does not bypass mTLS, modify JVM-wide defaults, or change other clients' TLS policies. Production must verify trust and matching SANs. This is distinct from the engine's southbound HTTP/JSON-RPC verification policy. +Connection and read timeouts both default to 30 seconds. Supply +`new LoadPsop.Timeouts(connectTimeout, readTimeout)` to set positive per-call values. ### RegistryClient @@ -629,7 +651,7 @@ not add private negotiation-state keys to the wire message. ## Thread Safety -- The engine client is thread-safe. Concurrent collections are used internally. +- The default engine client supports concurrent tasks. Each workflow invocation binds its own negotiation strategy and event callback without mutating client defaults. - `ControlPoint` implementations must be thread-safe if used from multiple workflow executions concurrently. - `EventCallback.onEvent` is called from multiple threads (main + SSE worker threads). Use synchronization if needed. diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 6406e4b0..c448b935 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -162,7 +162,7 @@ Each `next` entry is a `JumpCondition(step, condition)`; an empty condition make ```java WorkflowStep diagnosis = WorkflowStep.builder() .name("diagnosis") - .subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build())) + .subtasks(List.of(Task.builder().agent("Scheduled Agent A").description("diagnose").build())) .next(List.of( new JumpCondition("notify", ""), // unconditional: always active, never reaches onRoute new JumpCondition("hardware_recovery", "fault.hardware"), diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 40266480..4b1502b7 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -158,6 +158,10 @@ try { Required: `psop`, `controlPoint`. All other config items have defaults. +When the host supplies a configured `engineClient`, that client owns AgentCard, runtime, TLS, and credentials +construction. Do not also set the corresponding `ExecutePsop.Builder` options; conflicting settings fail before +execution instead of being silently ignored. + ## 5. Configuration ### 5.1 .env File @@ -429,6 +433,12 @@ transport/runtime/context independent of the task client and of one another. The complete ReceivedMessage, and closes on the host-defined terminal event. acknowledgement() and completion() separately represent ACK and actual stream exit; neither is a workflow prerequisite. +`heartbeat().lastEventAt()` is the last transport activity. A standard SSE comment heartbeat such as `: heartbeat` +refreshes it without increasing `eventCount`, which counts decoded A2A business events only. Use +`lastBusinessEventAt()` for the last business-message time. Cancelling the Future returned by `subscribeToTask` +only stops the local wait; close `NotificationSubscription` for an independent subscription, or close its owning +client/runtime for a client-managed task subscription. + `WorkflowEngineClientConfig.notificationAckTimeoutSeconds` controls the initial subscription ACK wait and defaults to 300 seconds. The Spring sample exposes it as `a2a.notification-ack-timeout-seconds` and `A2A_NOTIFICATION_ACK_TIMEOUT_SECONDS`. It does not control SSE idle time after acknowledgement. diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 46aa4d81..1d900ed6 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -26,7 +26,7 @@ |------------------------------------------|------|-------------|---------------------------------| | `psop(Workflow)` | 必填 | - | PSOP 工作流定义 | | `agentCards(List)` | 可选 | `List.of()` | 被调度智能体的 AgentCard；存在远程步骤且未传入已配置 `engineClient` 时必须提供 | -| `engineClient(WorkflowEngineClient)` | 可选 | null | 预配置客户端（null=自动创建） | +| `engineClient(WorkflowEngineClient)` | 可选 | null | 预配置客户端（null=自动创建）；传入后不可再设置 AgentCard、runtime、TLS 或 credentials 构造参数 | | `controlPoint(ControlPoint)` | 必填 | - | 用户决策实现 | | `runtimeIntent(String)` | 可选 | `""` | 自然语言意图，用于上下文组装 | | `lang(String)` | 可选 | `"zh"` | 语言提示（`"zh"` 或 `"en"`） | @@ -61,6 +61,8 @@ ExecutionResult result = ExecutePsop.builder() CompletableFuture sendTask(String agentName, MessageContent content); CompletableFuture sendTask(String agentName, MessageContent content, NegotiationStrategy negotiationStrategy); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy, EventCallback invocationEvents); CompletableFuture getTask(String agentName, String taskId); CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); @@ -74,6 +76,8 @@ void close(); `listTasks` 与 `cancelTask` 通过消息下发使用的同一 AgentCard、认证配置和自定义运行时暴露标准 A2A 任务管理操作。查询结果受认证身份权限约束，只包含当前身份可见的任务；调用方必须使用 `nextPageToken` 完成分页。关闭本地客户端或通知流不等同于取消远端任务。 +当前 `subscribeToTask` 返回的 Future 表示订阅结果，不是底层 SSE 的独立关闭句柄；取消 Future +不保证终止传输。非终态订阅需要通过 owning Client/Runtime 的 `close()` 释放。 `sendTask` 在 DAG 外执行宿主提供的最终任务内容。每次调用使用新的 context，并复用工作流任务的发送、等待、查询和 Negotiation-T 闭环。重载方法接受本次调用独有的协商策略，不修改客户端级 ControlPoint。超时、缺少处理器或其他本地交互失败后， @@ -107,6 +111,11 @@ NotificationSubscription openNotification(String agentName, MessageContent conte 授权和订阅接收宿主生成的最终内容。三类操作使用独立 transport/runtime/context，成功与否不阻断工作流。openNotification 先注册 handle 再开始 I/O，监听器直接收到 handle 和完整 ReceivedMessage。acknowledgement() 是实际 ACK；超时失败。close() 请求关闭，completion() 在流真正退出后完成。 +`heartbeat().lastEventAt()` 为兼容保留旧名称，实际表示最后一次传输活动；标准 SSE `: heartbeat` +会刷新它，但不会增加只统计已解码 A2A 业务事件的 `eventCount`。`lastBusinessEventAt()` 返回最近业务事件时间。 + +`new DefaultExtensionSender(transport)` 默认拥有并在 `close()` 时关闭 transport；仅当 transport +由其他组件统一管理时使用 `DefaultExtensionSender.nonOwning(transport)`，并由调用方关闭订阅和 transport。 ### WorkflowEngineClientConfig @@ -131,6 +140,8 @@ handle 再开始 I/O，监听器直接收到 handle 和完整 ReceivedMessage。 | `credentialsConfig` | `Map` | null | 内联凭证配置；AgentCard 声明安全要求时必须匹配 | | `maxNegotiationExchanges` | `int` | `3` | 本地交互预算，独立于 SDK maxRounds；耗尽仅本地失败 | +`credentialsConfigPath` 与 `credentialsConfig` 是两种互斥的内置凭据来源，同时设置会在构建配置时失败。 + ```java WorkflowEngineClientConfig config = WorkflowEngineClientConfig.builder() .sslVerify(true) @@ -259,7 +270,7 @@ Map normalized = AgentCardNormalizer.normalize(rawMap); | `A2ATExtension` | 规范扩展名称与 URI | | `A2ATransport` | 底层传输、认证、响应组装与订阅生命周期 | | `DefaultWorkflowEngineClient` | `WorkflowEngineClient` 默认实现 | -| `DefaultExtensionSender` | 基于 `A2ATransport` 的 `ExtensionSender` 默认实现 | +| `DefaultExtensionSender` | 基于 `A2ATransport` 的 `ExtensionSender` 默认实现；公开构造器拥有 transport，`nonOwning` 工厂不拥有 | | `DefaultA2AJavaClientRuntime` | HTTP/JSON-RPC/gRPC 的默认 A2A Java SDK runtime | | `CredentialCrypto` | AES-GCM 凭据加密工具与命令行入口 | | `EnvFileLoader` | 显式加载宿主自有 `.env` 配置 | @@ -343,6 +354,8 @@ public class EventCallback { ```java static Workflow load(String baseUrl, String psopId, String accessToken, boolean sslVerify) +static Workflow load(String baseUrl, String psopId, + String accessToken, boolean sslVerify, LoadPsop.Timeouts timeouts) static Workflow load(String baseUrl, String psopId) ``` @@ -355,6 +368,9 @@ GET `/api/v1/orchestrate/psop/{psop_id}`。返回完整工作流（含步骤、 static List search( String baseUrl, String intent, int topN, String accessToken, boolean sslVerify) +static List search( + String baseUrl, String intent, int topN, + String accessToken, boolean sslVerify, LoadPsop.Timeouts timeouts) static List search( String baseUrl, String intent) @@ -366,6 +382,7 @@ LoadPsop 的简便重载默认 `sslVerify=true`，使用 JVM 信任库并校验 显式传 `false` 时，仅该编排中心 HTTPS 连接跳过证书链和主机名校验，可免配本地 CA 文件联调。 服务端仍须提供 HTTPS 证书；此开关不绕过 mTLS，也不修改其他客户端或 JVM 的全局 TLS 策略。 生产环境必须保持验证并配置正确的服务端 SAN 与信任库；本设置不是引擎南向 HTTP/JSON-RPC 的 TLS 策略变更。 +连接与读取超时默认均为 30 秒；可通过 `new LoadPsop.Timeouts(connectTimeout, readTimeout)` 为单次调用设置正值。 ### RegistryClient @@ -592,7 +609,7 @@ getReceivedMessages() 是保留层级的响应来源，getOutputs() 为便利投 ## 线程安全 -- 引擎客户端线程安全，内部使用并发集合。 +- 默认引擎客户端可并发执行任务；工作流为每次调用绑定独立协商策略和事件回调，不改写客户端默认回调。 - `ControlPoint` 实现若在多工作流并发执行中使用，需自行保证线程安全。 - `EventCallback.onEvent` 从多个线程调用（主线程 + SSE 工作线程），需要时使用同步。 diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 282bf59f..1562d874 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -139,7 +139,7 @@ RouteDecision.allow (reason) 激活该边，或返回 RouteDecision.deny (reason ```java WorkflowStep diagnosis = WorkflowStep.builder() .name("diagnosis") - .subtasks(List.of(Task.builder().agent("SPN Domain Agent City1").description("diagnose").build())) + .subtasks(List.of(Task.builder().agent("被调度智能体 A").description("diagnose").build())) .next(List.of( new JumpCondition("notify", ""), // 无条件：始终激活，不经 onRoute new JumpCondition("hardware_recovery", "fault.hardware"), diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 1df47202..fd956522 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -154,6 +154,9 @@ try { 必填项：`psop`、`controlPoint`。其余配置项都有默认值。 +若宿主传入已配置好的 `engineClient`，AgentCard、runtime、TLS 和 credentials 均由该 client +负责，不要再设置对应的 `ExecutePsop.Builder` 构建参数；混用会在启动执行前直接报错，避免配置被静默忽略。 + ## 5. 配置 ### 5.1 .env 文件 @@ -407,6 +410,11 @@ NotificationSubscription openNotification(String agentName, MessageContent conte transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage，在宿主定义的终态事件上关闭。 handle.acknowledgement() 和 completion() 分别表示 ACK 和真实流退出，两者都不是工作流前提。 +`heartbeat().lastEventAt()` 表示最后一次传输活动；标准 SSE comment 心跳（如 `: heartbeat`）会刷新该时间， +但不会增加只统计已解码 A2A 业务事件的 `eventCount`。需要判断最后一条业务消息时间时使用 +`lastBusinessEventAt()`。取消 `subscribeToTask` 返回的 Future 只停止本地等待；结束独立订阅应关闭 +`NotificationSubscription`，结束客户端管理的任务订阅则关闭其 client/runtime。 + `WorkflowEngineClientConfig.notificationAckTimeoutSeconds` 控制首次订阅 ACK 的等待时间，默认 300 秒；Spring 示例可通过 `a2a.notification-ack-timeout-seconds` 或环境变量 `A2A_NOTIFICATION_ACK_TIMEOUT_SECONDS` 覆盖。该参数不控制 ACK 后的 SSE 空闲时间。