From 984ed2cde935723a046e28ac4bb1c6b1984d0af3 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Fri, 11 Sep 2026 18:21:14 +0800 Subject: [PATCH 1/3] feat(client): add one-shot task dispatch to ExtensionSender sendTask(agentName, content) dispatches a single task without a workflow: the host supplies final content (structured Task-T prompt or plain text), the engine envelopes, authenticates, sends and returns the complete SendMessageResult. No ControlPoint is invoked and no negotiation is attempted -- INPUT_REQUIRED is returned to the caller as-is. Each call uses a fresh context independent of any workflow session, matching sendAuthorization's isolation model. This lets hosts mix workflow orchestration with direct single-agent calls using the same transport, credentials and protocol logging. Signed-off-by: Zhoujie628 --- CHANGELOG.md | 5 ++ docs/en/API_REFERENCE.md | 7 ++ docs/en/INTEGRATION_GUIDE.md | 1 + docs/zh/API_REFERENCE.md | 1 + docs/zh/INTEGRATION_GUIDE.md | 1 + .../engine/client/DefaultExtensionSender.java | 12 ++++ .../engine/client/ExtensionSender.java | 12 ++++ .../engine/client/ExtensionSenderTest.java | 71 +++++++++++++++++++ 8 files changed, 110 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 207d5fc6..a4346537 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,11 @@ ## [Unreleased] +- Add `ExtensionSender.sendTask(agentName, content)`: a one-shot task dispatch independent of any + workflow. The host supplies final content; the engine envelopes, authenticates, sends and returns + the complete `SendMessageResult`. No `ControlPoint` is invoked and no negotiation is attempted -- + `INPUT_REQUIRED` is returned to the caller as-is. Each call uses a fresh context, matching + `sendAuthorization`. - Increase the default Notification-T acknowledgement timeout from 5 seconds to 5 minutes and expose the sample Spring setting as `a2a.notification-ack-timeout-seconds` / `A2A_NOTIFICATION_ACK_TIMEOUT_SECONDS`. diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index c9a33aee..e30708b6 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -97,10 +97,17 @@ task success, even if the dispatched agent acknowledges it with COMPLETED. ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` +`sendTask` dispatches a one-shot task without a workflow: the host supplies final content (structured +Task-T prompt or plain text), the engine envelopes, authenticates and sends it, and returns the +complete `SendMessageResult`. No `ControlPoint` is invoked and no negotiation is attempted -- if the +agent responds with `INPUT_REQUIRED`, the result carries that state and the caller decides the next +step. Each call uses a fresh context independent of any workflow session. + Authorization and notification accept host-generated final content. Use separate transport/runtime/context instances; 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; diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 2a83834b..2d8fa8b9 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -402,6 +402,7 @@ task success, even if the dispatched agent acknowledges it with COMPLETED. ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 20ffae34..d069eef1 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -92,6 +92,7 @@ COMPLETED 确认，也不能判为任务成功。 ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index bb94b6e5..72bc0ccb 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -384,6 +384,7 @@ COMPLETED 确认，也不能判为任务成功。 ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` 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..df4a1c82 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 @@ -49,6 +49,18 @@ public CompletableFuture sendAuthorization( } } + @Override + public CompletableFuture sendTask( + String agentName, MessageContent content) { + Objects.requireNonNull(content, "content"); + AgentCard card = transport.getCard(agentName); + if (card == null) throw new IllegalArgumentException("Agent not found: " + agentName); + String contextId = UUID.randomUUID().toString(); + return transport + .send(card, agentName, content, contextId, null, null) + .whenComplete((result, error) -> transport.closeConversation(card, contextId)); + } + @Override public NotificationSubscription openNotification( String agentName, diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java index 097c741e..af58d22b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java @@ -30,6 +30,18 @@ public interface ExtensionSender { /** Sends final authorization content on this sender's independent transport. */ CompletableFuture sendAuthorization(String agentName, MessageContent content); + /** + * Sends a one-shot task to an agent without a workflow context. + * + *

The host supplies final content (structured Task-T prompt or plain text); the engine + * envelopes it, authenticates, sends, and assembles the complete response. No {@code + * ControlPoint} is invoked and no negotiation is attempted: if the agent responds with + * {@code INPUT_REQUIRED}, the result carries that state and the caller decides the next step. + * The operation uses a fresh context independent of any workflow session, exactly like + * {@link #sendAuthorization(String, MessageContent)}. + */ + CompletableFuture sendTask(String agentName, MessageContent content); + /** Registers a handle before starting I/O; early callbacks may close it directly. */ NotificationSubscription openNotification( String agentName, 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..f99ded30 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 @@ -80,6 +80,77 @@ void sendsFinalAuthorizationWithoutGeneratingOrSharingContext() throws Exception } } + @Test + void sendsOneShotTaskWithFreshContextAndNoControlPoint() throws Exception { + List contexts = new CopyOnWriteArrayList<>(); + List taskTexts = new CopyOnWriteArrayList<>(); + try (var transport = + new A2ATransport( + List.of(card()), + DefaultWorkflowEngineClientNegotiationTest.runtime( + params -> { + contexts.add(params.message().contextId()); + params + .message() + .parts() + .forEach( + part -> { + if (part instanceof TextPart textPart) taskTexts.add(textPart.text()); + }); + return List.of( + DefaultWorkflowEngineClientNegotiationTest.response( + params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("task-done"))); + }), + WorkflowEngineClientConfig.builder().build())) { + var sender = new DefaultExtensionSender(transport); + + var result1 = sender.sendTask("test", MessageContent.text("diagnose port")).join(); + sender.sendTask("test", MessageContent.text("diagnose port")).join(); + + assertEquals("TASK_STATE_COMPLETED", result1.getTaskState()); + assertNotEquals(contexts.get(0), contexts.get(1)); + assertEquals(2, taskTexts.size()); + assertEquals("diagnose port", taskTexts.get(0)); + assertEquals("diagnose port", taskTexts.get(1)); + } + } + + @Test + void oneShotTaskReturnsInputRequiredResultWithoutNegotiation() throws Exception { + try (var transport = + new A2ATransport( + List.of(card()), + DefaultWorkflowEngineClientNegotiationTest.runtime( + params -> + List.of( + DefaultWorkflowEngineClientNegotiationTest.response( + params, + TaskState.TASK_STATE_INPUT_REQUIRED, + MessageContent.text("need-more-info")))), + WorkflowEngineClientConfig.builder().build())) { + var result = + new DefaultExtensionSender(transport) + .sendTask("test", MessageContent.text("diagnose")) + .join(); + assertEquals("TASK_STATE_INPUT_REQUIRED", result.getTaskState()); + } + } + + @Test + void oneShotTaskFailsFastForUnknownAgent() throws Exception { + try (var transport = + new A2ATransport( + List.of(card()), + DefaultWorkflowEngineClientNegotiationTest.runtime(params -> List.of()), + WorkflowEngineClientConfig.builder().build())) { + var sender = new DefaultExtensionSender(transport); + assertThrows( + IllegalArgumentException.class, () -> sender.sendTask("unknown", MessageContent.text("x"))); + assertThrows( + NullPointerException.class, () -> sender.sendTask("test", null)); + } + } + @Test void earlyEventCanCloseHandleAndCompletionWaitsForActualExit() throws Exception { CountDownLatch callbackRan = new CountDownLatch(1); From 5b5edf4652dd2111b408051faeb9444e1a5f559a Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Sat, 12 Sep 2026 10:36:03 +0800 Subject: [PATCH 2/3] refactor(client): move standalone tasks to workflow client Signed-off-by: Zhoujie628 --- .../examples/demo/A2ATProtocolCases.java | 19 +- .../engine/examples/demo/SpringSpnDemo.java | 2 +- .../examples/demo/HostQuickStartTest.java | 2 +- .../server/EmbeddedA2AServerTest.java | 39 ++- .../workbench/StaleTaskCleanerTest.java | 11 +- .../engine/client/A2AJavaClientRuntime.java | 6 +- .../workflow/engine/client/A2ATExtension.java | 2 +- .../engine/client/DefaultExtensionSender.java | 12 - .../client/DefaultWorkflowEngineClient.java | 173 +++++++++- .../engine/client/ExtensionSender.java | 12 - .../engine/client/WorkflowEngineClient.java | 18 +- .../engine/core/WorkflowExecutor.java | 26 +- .../engine/StubWorkflowEngineClient.java | 18 +- .../engine/client/A2ATransportHeaderTest.java | 2 +- ...ltWorkflowEngineClientNegotiationTest.java | 317 +++++++++++++++++- .../engine/client/ExtensionSenderTest.java | 71 ---- .../core/RemoteA2AErrorFailureTest.java | 14 +- .../engine/core/WorkflowExecutorTest.java | 68 ++++ 18 files changed, 642 insertions(+), 170 deletions(-) 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 d834b893..dd964959 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 @@ -250,11 +250,20 @@ private CompletableFuture sendDiagnosis( dev.openan.workflow.engine.model.BusinessInput.data( SpnCasePrompts.privateLineComplaintData())) .build(); - var callbacks = - new dev.openan.workflow.engine.control.DefaultControlPoint( - new dev.openan.workflow.engine.examples.negotiation.NegotiationStrategy( - EnvResolver.resolveEnvPath())); - return client.dispatch(request, content, callbacks); + var strategy = + new dev.openan.workflow.engine.examples.negotiation.NegotiationStrategy( + EnvResolver.resolveEnvPath()); + return client.sendTask( + AGENT_NAME, + content, + negotiation -> + strategy.resolve( + new dev.openan.workflow.engine.model.NegotiationRequest( + request, + negotiation.originalSubmission(), + negotiation.received(), + negotiation.previousExchanges(), + negotiation.remainingWait()))); } private void case_7_2() throws Exception { diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemo.java b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemo.java index 658daac0..1b85eab9 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemo.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemo.java @@ -241,7 +241,7 @@ private SendMessageResult sendTaskToWorkbench() throws Exception { SpnCasePrompts.TASK_TEXT.length()); SendMessageResult result = client - .sendMessage( + .sendTask( WB_AGENT_NAME, dev.openan.workflow.engine.client.A2atMessages.from( dev.openan.workflow.engine.examples.util.A2ATInitialization.create( 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 4d3c73d9..b7cd2f5d 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,7 @@ 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("dispatch")) { + if (method.getName().equals("sendTask") && method.getParameterCount() == 3) { 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/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 0ee41902..f2d82609 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 @@ -130,6 +130,13 @@ void setUp() throws Exception { false, "extensions", List.of( + Map.of( + "uri", + A2ATExtension.TASK_T.uri(), + "description", + "Task-T test extension", + "required", + false), Map.of( "uri", "https://projects.tmforum.org/a2aproject" @@ -297,12 +304,21 @@ void missingTaskSubscriptionReturnsAStandardA2AErrorBeforeSseStarts() throws Exc @Test void testSendMessage() throws Exception { + var request = request(); + var callbacks = new WorkbenchControlPoint(sdkEnvPath); SendMessageResult result = client - .dispatch( - request(), - new WorkbenchControlPoint(sdkEnvPath).onTask(request()).join(), - new WorkbenchControlPoint(sdkEnvPath)) + .sendTask( + request.getAgentName(), + callbacks.onTask(request).join(), + negotiation -> + callbacks.onNegotiation( + new dev.openan.workflow.engine.model.NegotiationRequest( + request, + negotiation.originalSubmission(), + negotiation.received(), + negotiation.previousExchanges(), + negotiation.remainingWait()))) .join(); assertNotNull(result); assertFalse( @@ -334,15 +350,24 @@ void testNegotiationAutoLoopCompletes() { taskPrompt); metadata.put( MetadataContent.TEMPLATE_URI_METADATA_KEY, StandardTemplates.PRIVATE_LINE_COMPLAINT.uri()); + var request = request(); + var callbacks = new WorkbenchControlPoint(sdkEnvPath); SendMessageResult result = client - .dispatch( - request(), + .sendTask( + request.getAgentName(), new dev.openan.workflow.engine.model.MessageContent( List.of(new org.a2aproject.sdk.spec.TextPart("diagnose SPN fault")), metadata, java.util.Set.of(A2ATExtension.TASK_T.uri())), - new WorkbenchControlPoint(sdkEnvPath)) + negotiation -> + callbacks.onNegotiation( + new dev.openan.workflow.engine.model.NegotiationRequest( + request, + negotiation.originalSubmission(), + negotiation.received(), + negotiation.previousExchanges(), + negotiation.remainingWait()))) .orTimeout(60, java.util.concurrent.TimeUnit.SECONDS) .join(); assertNotNull(result); diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java index 73e18c08..57a78ca9 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/StaleTaskCleanerTest.java @@ -25,7 +25,6 @@ import dev.openan.workflow.engine.control.EventCallback; import dev.openan.workflow.engine.model.MessageContent; import dev.openan.workflow.engine.model.SendMessageResult; -import dev.openan.workflow.engine.model.TaskRequest; import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -111,14 +110,16 @@ private static SendMessageResult result(Task task) { } @Override - public CompletableFuture dispatch( - TaskRequest request, MessageContent content, ControlPoint callbacks) { + public CompletableFuture sendTask( + String agentName, MessageContent content) { throw new UnsupportedOperationException(); } @Override - public CompletableFuture sendMessage( - String agentName, MessageContent content) { + public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { throw new UnsupportedOperationException(); } 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 49b9f747..a81bcfaa 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 @@ -37,8 +37,8 @@ * when one is provided (or auto-created). This avoids hand-written HTTP code and reuses the SDK's * transport, SSE parsing, and error handling. * - *

Mirrors {@code A2AJavaClientRuntime} from the a2a-t-sdk-java sample module, but lives in the - * engine package so the engine can depend on it directly. + *

Custom implementations can replace the wire channel while keeping the engine's task state + * machine and authentication contract unchanged. */ public interface A2AJavaClientRuntime { String CHANNEL_STATE_KEY = "dev.openan.workflow.engine.channel"; @@ -47,7 +47,7 @@ public interface A2AJavaClientRuntime { /** * Send a message to an agent and collect the streaming events. * - * @param agentCard the target agent's card as a map (from config or registry) + * @param agentCard the target agent's card (from configuration or a registry) * @param params the message send parameters (message, context, metadata) * @param callContext client call context with auth/extension headers * @param eventSink optional callback invoked for each intermediate event (status updates, diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java index df866f96..9b0dd486 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2ATExtension.java @@ -30,7 +30,7 @@ *

The engine handles these extensions automatically: * *

    - *
  • {@link #TASK_T} - host-generated task content (in-workflow) + *
  • {@link #TASK_T} - host-generated task content (workflow or standalone task) *
  • {@link #NEGOTIATION_T} - negotiation auto-loop (in-workflow) *
  • {@link #AUTHORIZATION_T} - independent whitelist authorization operation *
  • {@link #NOTIFICATION_T} - independent result subscription 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 df4a1c82..a3ac05c7 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 @@ -49,18 +49,6 @@ public CompletableFuture sendAuthorization( } } - @Override - public CompletableFuture sendTask( - String agentName, MessageContent content) { - Objects.requireNonNull(content, "content"); - AgentCard card = transport.getCard(agentName); - if (card == null) throw new IllegalArgumentException("Agent not found: " + agentName); - String contextId = UUID.randomUUID().toString(); - return transport - .send(card, agentName, content, contextId, null, null) - .whenComplete((result, error) -> transport.closeConversation(card, contextId)); - } - @Override public NotificationSubscription openNotification( String agentName, 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 305bdedb..1a8fdfa4 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 @@ -22,11 +22,15 @@ import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.control.EventCallback; import dev.openan.workflow.engine.control.EventType; +import dev.openan.workflow.engine.control.NegotiationStrategy; import dev.openan.workflow.engine.model.*; import java.time.Duration; import java.util.*; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import net.openan.a2at.sdk.core.model.NegotiationContext; import net.openan.a2at.sdk.core.model.NegotiationPerformative; @@ -133,7 +137,7 @@ public long callbackTimeoutSeconds() { } @Override - public CompletableFuture sendMessage( + public CompletableFuture sendTask( String agentName, MessageContent content) { TaskRequest request = TaskRequest.builder() @@ -141,14 +145,35 @@ public CompletableFuture sendMessage( .executionId(UUID.randomUUID().toString()) .taskId(UUID.randomUUID().toString()) .build(); - return dispatch(request, content, controlPoint); + return dispatchTask(request, content, controlPoint); } @Override - public CompletableFuture dispatch( + public CompletableFuture sendTask( + String agentName, MessageContent content, NegotiationStrategy negotiationStrategy) { + if (negotiationStrategy == null) { + return CompletableFuture.failedFuture(new NullPointerException("negotiationStrategy")); + } + TaskRequest request = + TaskRequest.builder() + .agentName(agentName) + .executionId(UUID.randomUUID().toString()) + .taskId(UUID.randomUUID().toString()) + .build(); + ControlPoint callbacks = + ControlPoint.builder().onNegotiation(negotiationStrategy::resolve).build(); + return dispatchTask(request, content, callbacks); + } + + private CompletableFuture dispatchTask( TaskRequest request, MessageContent content, ControlPoint callbacks) { - Objects.requireNonNull(request, "request"); - Objects.requireNonNull(content, "onTask returned null content"); + if (request == null) { + return CompletableFuture.failedFuture(new NullPointerException("request")); + } + if (content == null) { + return CompletableFuture.failedFuture( + new NullPointerException("onTask returned null content")); + } AgentCard card = transport.getCard(request.getAgentName()); if (closed.get() || card == null) return CompletableFuture.failedFuture( @@ -156,13 +181,37 @@ public CompletableFuture dispatch( closed.get() ? "Workflow client closed" : "Agent not found: " + request.getAgentName())); + try { + validateTaskExtension(card, content); + } catch (RuntimeException error) { + return CompletableFuture.failedFuture(error); + } Invocation invocation = new Invocation(request, content, callbacks); invocations.add(invocation); - invocation.completion.orTimeout(callbackTimeoutSeconds(), TimeUnit.SECONDS); + CompletableFuture.delayedExecutor(callbackTimeoutSeconds(), TimeUnit.SECONDS) + .execute( + () -> { + if (!invocation.completion.isDone()) { + completeExceptionallyAfterCleanup( + card, + invocation, + new TimeoutException( + "Task interaction timed out after " + + callbackTimeoutSeconds() + + " seconds")); + } + }); invocation.completion.whenComplete( (result, error) -> { invocations.remove(invocation); - transport.closeConversation(card, invocation.contextId); + if (invocation.completion.isCancelled()) { + cancelRemoteTask(card, invocation, null) + .whenComplete( + (ignored, cleanupError) -> + transport.closeConversation(card, invocation.contextId)); + } else { + transport.closeConversation(card, invocation.contextId); + } if (error != null && invocation.exchanges > 0) emit( EventType.NEGOTIATION_FAILED, @@ -179,12 +228,84 @@ public CompletableFuture dispatch( .thenCompose(result -> advance(card, invocation, result)) .whenComplete( (result, error) -> { - if (error != null) invocation.completion.completeExceptionally(error); - else invocation.completion.complete(result); + if (error != null) completeExceptionallyAfterCleanup(card, invocation, error); + else if (invocation.finishing.compareAndSet(false, true)) + invocation.completion.complete(result); }); return invocation.completion; } + private static void validateTaskExtension(AgentCard card, MessageContent content) { + String taskExtension = A2ATExtension.TASK_T.uri(); + boolean activated = content.extensions().contains(taskExtension); + boolean supplied = content.metadata().containsKey(taskExtension); + if (!activated && !supplied) return; + if (!activated || !supplied) { + throw new IllegalArgumentException( + "Task-T content must contain matching extension activation and metadata"); + } + if (!A2ATransport.extractExtensionUris(card).contains(taskExtension)) { + throw new IllegalArgumentException("Target agent does not declare Task-T: " + card.name()); + } + } + + private void completeExceptionallyAfterCleanup( + AgentCard card, Invocation invocation, Throwable error) { + if (!invocation.finishing.compareAndSet(false, true)) { + // A timeout may win before the first send exposes its remote task id. If that result arrives + // later, advance() records the id and reaches this path again so the orphan can still be + // cancelled. cancelRemoteTask() deduplicates an already-started cleanup. + cancelRemoteTask(card, invocation, null); + return; + } + Throwable primary = unwrap(error); + cancelRemoteTask(card, invocation, primary) + .whenComplete((ignored, cleanupError) -> invocation.completion.completeExceptionally(primary)); + } + + private CompletableFuture cancelRemoteTask( + AgentCard card, Invocation invocation, Throwable primary) { + String remoteTaskId = invocation.remoteTaskId; + if (remoteTaskId == null || invocation.remoteTaskTerminal) { + return CompletableFuture.completedFuture(null); + } + synchronized (invocation) { + if (invocation.cleanup != null) return invocation.cleanup; + log.info( + "[EngineClient] Canceling non-final remote task after local interaction ended: agent={}, taskId={}", + invocation.task.getAgentName(), + remoteTaskId); + invocation.cleanup = + transport + .cancelTask(card, invocation.task.getAgentName(), remoteTaskId) + .handle( + (result, cleanupError) -> { + if (cleanupError == null) { + invocation.remoteTaskTerminal = true; + } else { + Throwable failure = unwrap(cleanupError); + if (primary != null && failure != primary) primary.addSuppressed(failure); + log.warn( + "[EngineClient] Failed to cancel non-final remote task: agent={}, taskId={}", + invocation.task.getAgentName(), + remoteTaskId, + failure); + } + return null; + }); + return invocation.cleanup; + } + } + + private static Throwable unwrap(Throwable error) { + Throwable current = error; + while ((current instanceof CompletionException || current instanceof ExecutionException) + && current.getCause() != null) { + current = current.getCause(); + } + return current; + } + private CompletableFuture send( AgentCard card, Invocation invocation, MessageContent content, String taskId) { if (!invocation.active()) @@ -208,11 +329,11 @@ private CompletableFuture send( } private Map invocationTrace(Invocation invocation) { - Map trace = new HashMap<>(); + Map trace = new HashMap<>(invocation.parentTrace); if (invocation.task.getExecutionId() != null) - trace.put("executionId", invocation.task.getExecutionId()); + trace.putIfAbsent("executionId", invocation.task.getExecutionId()); if (invocation.task.getTaskId() != null) - trace.put("logicalTaskId", invocation.task.getTaskId()); + trace.putIfAbsent("logicalTaskId", invocation.task.getTaskId()); trace.put("attempt", invocation.attempt); trace.put("contextId", invocation.contextId); return trace; @@ -220,9 +341,6 @@ private Map invocationTrace(Invocation invocation) { private CompletableFuture advance( AgentCard card, Invocation invocation, SendMessageResult result) { - if (!invocation.active()) - return CompletableFuture.failedFuture( - new java.util.concurrent.CancellationException("Task interaction is no longer active")); if (result.getTask() != null) { if (result.getTask().id() == null || result.getTask().id().isBlank() @@ -233,11 +351,25 @@ private CompletableFuture advance( new IllegalArgumentException("Remote task/context identity changed")); } invocation.remoteTaskId = result.getTask().id(); + invocation.remoteTaskTerminal = + result.getTask().status() != null + && result.getTask().status().state() != null + && result.getTask().status().state().isFinal(); } + if (!invocation.active()) + return CompletableFuture.failedFuture( + new java.util.concurrent.CancellationException("Task interaction is no longer active")); if ("TASK_STATE_SUBMITTED".equals(result.getTaskState()) || "TASK_STATE_WORKING".equals(result.getTaskState())) { return observeTask(card, invocation).thenCompose(next -> advance(card, invocation, next)); } + if (!"TASK_STATE_INPUT_REQUIRED".equals(result.getTaskState()) + && result.getTask() != null + && !invocation.remoteTaskTerminal) { + return CompletableFuture.failedFuture( + new IllegalStateException( + "Unsupported non-final remote task state: " + result.getTaskState())); + } if (!"TASK_STATE_INPUT_REQUIRED".equals(result.getTaskState())) { emitAgentResponse(invocation.task.getAgentName(), result); return CompletableFuture.completedFuture(result); @@ -482,12 +614,16 @@ private final class Invocation { final ControlPoint callbacks; final String contextId = UUID.randomUUID().toString(); final String attempt = UUID.randomUUID().toString(); + final Map parentTrace = Map.copyOf(WireLog.context()); final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(callbackTimeoutSeconds()); final CompletableFuture completion = new CompletableFuture<>(); final Map> history = new HashMap<>(); final Set answered = new HashSet<>(); final Map contexts = new HashMap<>(); - String remoteTaskId; + final AtomicBoolean finishing = new AtomicBoolean(); + volatile String remoteTaskId; + volatile boolean remoteTaskTerminal; + CompletableFuture cleanup; int exchanges; Invocation(TaskRequest task, MessageContent content, ControlPoint callbacks) { @@ -501,7 +637,10 @@ Duration remaining() { } boolean active() { - return !closed.get() && !completion.isDone() && !remaining().isZero(); + return !closed.get() + && !finishing.get() + && !completion.isDone() + && !remaining().isZero(); } } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java index af58d22b..097c741e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ExtensionSender.java @@ -30,18 +30,6 @@ public interface ExtensionSender { /** Sends final authorization content on this sender's independent transport. */ CompletableFuture sendAuthorization(String agentName, MessageContent content); - /** - * Sends a one-shot task to an agent without a workflow context. - * - *

    The host supplies final content (structured Task-T prompt or plain text); the engine - * envelopes it, authenticates, sends, and assembles the complete response. No {@code - * ControlPoint} is invoked and no negotiation is attempted: if the agent responds with - * {@code INPUT_REQUIRED}, the result carries that state and the caller decides the next step. - * The operation uses a fresh context independent of any workflow session, exactly like - * {@link #sendAuthorization(String, MessageContent)}. - */ - CompletableFuture sendTask(String agentName, MessageContent content); - /** Registers a handle before starting I/O; early callbacks may close it directly. */ NotificationSubscription openNotification( String agentName, 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 20e4d06b..45bca8a7 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 @@ -21,9 +21,9 @@ import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.control.EventCallback; +import dev.openan.workflow.engine.control.NegotiationStrategy; import dev.openan.workflow.engine.model.MessageContent; import dev.openan.workflow.engine.model.SendMessageResult; -import dev.openan.workflow.engine.model.TaskRequest; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; @@ -32,12 +32,18 @@ /** Sends final content and coordinates task interaction. Content generation belongs to the host. */ public interface WorkflowEngineClient { - /** Dispatches a prepared workflow activation; target and protocol association remain internal. */ - CompletableFuture dispatch( - TaskRequest request, MessageContent content, ControlPoint callbacks); + /** + * Sends final task content outside the DAG using the configured negotiation callback. The same + * task/context association is retained until the remote task reaches a final state. + */ + CompletableFuture sendTask(String agentName, MessageContent content); - /** Sends final content outside the DAG using explicitly configured interaction callbacks. */ - CompletableFuture sendMessage(String agentName, MessageContent content); + /** + * Sends final task content outside the DAG with a per-call negotiation strategy. This overload + * does not mutate the client's configured {@link ControlPoint}. + */ + CompletableFuture sendTask( + String agentName, MessageContent content, NegotiationStrategy negotiationStrategy); /** * Maximum wait for one task interaction, from initial content preparation through remote task 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 15ea3604..c0acf141 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 @@ -20,11 +20,13 @@ package dev.openan.workflow.engine.core; import dev.openan.workflow.engine.client.WorkflowEngineClient; +import dev.openan.workflow.engine.client.WireLog; import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.control.EventCallback; import dev.openan.workflow.engine.control.EventType; import dev.openan.workflow.engine.model.ExecutionResult; import dev.openan.workflow.engine.model.JumpCondition; +import dev.openan.workflow.engine.model.NegotiationRequest; import dev.openan.workflow.engine.model.RouteDecision; import dev.openan.workflow.engine.model.RouteRequest; import dev.openan.workflow.engine.model.StepType; @@ -382,12 +384,26 @@ private CompletableFuture dispatchTask(WorkflowStep step, TaskReques return; } try { + Map trace = + Map.of( + "executionId", request.getExecutionId(), + "logicalTaskId", request.getTaskId()); var sent = - engineClient.dispatch( - request, - java.util.Objects.requireNonNull( - content, "onTask returned null content"), - controlPoint); + WireLog.call( + trace, + () -> + engineClient.sendTask( + request.getAgentName(), + java.util.Objects.requireNonNull( + content, "onTask returned null content"), + negotiation -> + controlPoint.onNegotiation( + new NegotiationRequest( + request, + negotiation.originalSubmission(), + negotiation.received(), + negotiation.previousExchanges(), + negotiation.remainingWait())))); result.whenComplete( (value, failure) -> { if (!sent.isDone()) sent.cancel(true); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java index 2d884c1c..d1c85eb0 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/StubWorkflowEngineClient.java @@ -70,15 +70,7 @@ public StubWorkflowEngineClient withDefaultTaskState(String state) { } @Override - public CompletableFuture dispatch( - dev.openan.workflow.engine.model.TaskRequest request, - dev.openan.workflow.engine.model.MessageContent content, - ControlPoint callbacks) { - return sendMessage(request.getAgentName(), content); - } - - @Override - public CompletableFuture sendMessage( + public CompletableFuture sendTask( String agentName, dev.openan.workflow.engine.model.MessageContent content) { String message = content.parts().stream() @@ -127,6 +119,14 @@ public CompletableFuture sendMessage( return CompletableFuture.completedFuture(result); } + @Override + public CompletableFuture sendTask( + String agentName, + dev.openan.workflow.engine.model.MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { + return sendTask(agentName, content); + } + @Override public void setControlPoint(ControlPoint controlPoint) { this.controlPoint = controlPoint; 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 6e91f25f..ce894f39 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 @@ -370,7 +370,7 @@ 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).sendMessage("Test Agent", content).join(); + new DefaultWorkflowEngineClient(transport).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 190ab874..777c0ed7 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 @@ -34,6 +34,16 @@ import org.junit.jupiter.api.Test; class DefaultWorkflowEngineClientNegotiationTest { + @Test + void publicClientContractExposesOnlyTheTaskOrientedSendName() { + Set methodNames = + Arrays.stream(WorkflowEngineClient.class.getMethods()) + .map(java.lang.reflect.Method::getName) + .collect(java.util.stream.Collectors.toSet()); + assertFalse(methodNames.contains("dispatch")); + assertFalse(methodNames.contains("sendMessage")); + } + static AgentCard card() throws Exception { return new com.fasterxml.jackson.databind.ObjectMapper() .registerModule(new AgentCardJacksonModule()) @@ -77,6 +87,7 @@ static TaskEvent response(MessageSendParams request, TaskState state, MessageCon } static A2AJavaClientRuntime runtime(Function> send) { + AtomicReference lastRequest = new AtomicReference<>(); return new A2AJavaClientRuntime() { public Iterable sendMessage( AgentCard card, @@ -84,15 +95,303 @@ public Iterable sendMessage( ClientCallContext context, Consumer sink, Consumer logs) { + lastRequest.set(params); List events = send.apply(params); if (sink != null) events.forEach(sink); return events; } + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + MessageSendParams request = lastRequest.get(); + if (request == null) throw new IllegalStateException("No task has been sent"); + return response(request, TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) + .getTask(); + } + public void close() {} }; } + @Test + void standaloneTaskUsesFreshContextAndConfiguredInteractionLoop() throws Exception { + List contexts = new CopyOnWriteArrayList<>(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime( + params -> { + contexts.add(params.message().contextId()); + return List.of( + response( + params, + TaskState.TASK_STATE_COMPLETED, + MessageContent.text("task-done"))); + }), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + assertEquals( + "TASK_STATE_COMPLETED", + client.sendTask("test", MessageContent.text("first")).join().getTaskState()); + assertEquals( + "TASK_STATE_COMPLETED", + client.sendTask("test", MessageContent.text("second")).join().getTaskState()); + } + assertEquals(2, contexts.size()); + assertNotEquals(contexts.get(0), contexts.get(1)); + } + + @Test + void standaloneTaskAcceptsPerCallNegotiationStrategy() throws Exception { + AtomicInteger sends = new AtomicInteger(); + AtomicInteger negotiations = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime( + params -> + List.of( + response( + params, + sends.incrementAndGet() == 1 + ? TaskState.TASK_STATE_INPUT_REQUIRED + : TaskState.TASK_STATE_COMPLETED, + sends.get() == 1 + ? negotiation("standalone", 1, NegotiationPerformative.PROPOSE) + : MessageContent.text("done")))), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + var result = + client + .sendTask( + "test", + MessageContent.text("start"), + request -> { + negotiations.incrementAndGet(); + return CompletableFuture.completedFuture( + new NegotiationReply.Send( + negotiation("standalone", 1, NegotiationPerformative.ACCEPT))); + }) + .join(); + + assertEquals("TASK_STATE_COMPLETED", result.getTaskState()); + assertEquals(1, negotiations.get()); + assertEquals(2, sends.get()); + } + } + + @Test + void standaloneTaskCancelsRemoteTaskBeforeFailingWhenNegotiationCannotContinue() + throws Exception { + AtomicInteger cancellations = new AtomicInteger(); + AtomicReference sent = new AtomicReference<>(); + A2AJavaClientRuntime runtime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + sent.set(params); + return List.of( + response( + params, + TaskState.TASK_STATE_INPUT_REQUIRED, + negotiation("standalone", 1, NegotiationPerformative.PROPOSE))); + } + + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + cancellations.incrementAndGet(); + assertEquals("remote-task", taskId); + return response( + sent.get(), TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) + .getTask(); + } + + public void close() {} + }; + try (var transport = + new A2ATransport( + List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + assertThrows( + CompletionException.class, + () -> client.sendTask("test", MessageContent.text("start")).join()); + assertEquals(1, cancellations.get()); + } + } + + @Test + void lateRemoteIdentityIsCanceledAfterLocalTimeout() throws Exception { + CountDownLatch canceled = new CountDownLatch(1); + AtomicReference sent = new AtomicReference<>(); + A2AJavaClientRuntime runtime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + sent.set(params); + java.util.concurrent.locks.LockSupport.parkNanos( + TimeUnit.MILLISECONDS.toNanos(1_200)); + return List.of( + response( + params, + TaskState.TASK_STATE_INPUT_REQUIRED, + negotiation("late", 1, NegotiationPerformative.PROPOSE))); + } + + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + canceled.countDown(); + return response( + sent.get(), TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) + .getTask(); + } + + public void close() {} + }; + WorkflowEngineClientConfig config = + WorkflowEngineClientConfig.builder().sendTimeoutSeconds(1).build(); + try (var transport = new A2ATransport(List.of(card()), runtime, config)) { + var client = new DefaultWorkflowEngineClient(transport, config); + + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", MessageContent.text("start")).join()); + assertInstanceOf(TimeoutException.class, error.getCause()); + assertTrue(canceled.await(3, TimeUnit.SECONDS)); + } + } + + @Test + void standaloneTaskCancelsUnsupportedAuthRequiredTask() throws Exception { + AtomicInteger cancellations = new AtomicInteger(); + AtomicReference sent = new AtomicReference<>(); + A2AJavaClientRuntime runtime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + sent.set(params); + return List.of( + response( + params, + TaskState.TASK_STATE_AUTH_REQUIRED, + MessageContent.text("authentication required"))); + } + + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + cancellations.incrementAndGet(); + return response( + sent.get(), TaskState.TASK_STATE_CANCELED, MessageContent.text("canceled")) + .getTask(); + } + + public void close() {} + }; + try (var transport = + new A2ATransport( + List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", MessageContent.text("start")).join()); + assertTrue(error.getCause().getMessage().contains("TASK_STATE_AUTH_REQUIRED")); + assertEquals(1, cancellations.get()); + } + } + + @Test + void standaloneTaskValidatesTaskTExtensionOnlyWhenActivated() throws Exception { + AtomicInteger sends = new AtomicInteger(); + try (var transport = + new A2ATransport( + List.of(card()), + runtime( + params -> { + sends.incrementAndGet(); + return List.of( + response( + params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done"))); + }), + WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + client.sendTask("test", MessageContent.text("plain task")).join(); + MessageContent taskT = + A2atMessages.from( + new MetadataContent( + "Task-T/example/v1", "structured task", A2ATExtension.TASK_T.uri()), + List.of(new TextPart("structured task"))); + CompletionException error = + assertThrows(CompletionException.class, () -> client.sendTask("test", taskT).join()); + assertTrue(error.getCause().getMessage().contains("does not declare Task-T")); + assertEquals(1, sends.get()); + } + } + + @Test + void standaloneTaskPreservesStandardA2AErrorWithoutCancelingAnUncreatedTask() + throws Exception { + AtomicInteger cancellations = new AtomicInteger(); + RemoteA2AErrorException remoteError = + RemoteA2AErrorException.fromPayload( + """ + {"error":{"code":429,"status":"RESOURCE_EXHAUSTED","message":"Task limit reached", + "details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo", + "reason":"TASK_LIMIT_REACHED","domain":"a2a-protocol.org"}]}} + """); + A2AJavaClientRuntime runtime = + new A2AJavaClientRuntime() { + public Iterable sendMessage( + AgentCard card, + MessageSendParams params, + ClientCallContext context, + Consumer sink, + Consumer logs) { + throw remoteError; + } + + public org.a2aproject.sdk.spec.Task cancelTask( + AgentCard card, String taskId, ClientCallContext context) { + cancellations.incrementAndGet(); + throw new AssertionError("No remote task was created"); + } + + public void close() {} + }; + try (var transport = + new A2ATransport( + List.of(card()), runtime, WorkflowEngineClientConfig.builder().build())) { + var client = new DefaultWorkflowEngineClient(transport); + + CompletionException error = + assertThrows( + CompletionException.class, + () -> client.sendTask("test", MessageContent.text("start")).join()); + RemoteA2AErrorException actual = RemoteA2AErrorException.findIn(error); + assertNotNull(actual); + assertEquals(429, actual.getHttpStatus()); + assertEquals("TASK_LIMIT_REACHED", actual.getReason()); + assertEquals(0, cancellations.get()); + } + } + @Test void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception { List sent = new CopyOnWriteArrayList<>(); @@ -126,7 +425,7 @@ void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception { .build()); assertEquals( "TASK_STATE_COMPLETED", - client.sendMessage("test", MessageContent.text("start")).join().getTaskState()); + client.sendTask("test", MessageContent.text("start")).join().getTaskState()); } assertEquals(2, sent.size()); assertNull(sent.get(0).message().taskId()); @@ -160,7 +459,7 @@ void terminalStateNeverRestartsOldNegotiation() throws Exception { return CompletableFuture.failedFuture(new AssertionError()); }) .build()); - client.sendMessage("test", MessageContent.text("start")).join(); + client.sendTask("test", MessageContent.text("start")).join(); } assertEquals(0, callbacks.get()); } @@ -193,7 +492,7 @@ void rejectsWrongContextNewProposeAndMissingCallbackWithoutSendingAgain() throws .build()); assertThrows( CompletionException.class, - () -> client.sendMessage("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", MessageContent.text("start")).join()); } assertEquals(1, sends.get()); } @@ -215,7 +514,7 @@ void plainInputRequiredIsExplicitlyUnsupported() throws Exception { var client = new DefaultWorkflowEngineClient(transport); assertThrows( CompletionException.class, - () -> client.sendMessage("test", MessageContent.text("start")).join()); + () -> client.sendTask("test", MessageContent.text("start")).join()); } } @@ -244,7 +543,7 @@ void abortAcknowledgementIsNotDiagnosisSuccessAndPreservesRemoteState() throws E new NegotiationReply.Send( negotiation("city1", 1, NegotiationPerformative.ABORT)))) .build()); - var result = client.sendMessage("test", MessageContent.text("start")).join(); + var result = client.sendTask("test", MessageContent.text("start")).join(); assertEquals("TASK_STATE_COMPLETED", result.getTaskState()); assertEquals("negotiation.aborted", result.getFailureCode()); } @@ -307,7 +606,7 @@ public void close() {} assertEquals( "TASK_STATE_COMPLETED", client - .sendMessage("test", MessageContent.text("start")) + .sendTask("test", MessageContent.text("start")) .get(5, TimeUnit.SECONDS) .getTaskState()); assertEquals(2, sends.get()); @@ -345,7 +644,7 @@ void canceledOrTimedOutNegotiationNeverSendsLateReply() throws Exception { return answer; }) .build()); - var result = client.sendMessage("test", MessageContent.text("start")); + var result = client.sendTask("test", MessageContent.text("start")); assertTrue(entered.await(1, TimeUnit.SECONDS)); if (timeout) assertThrows(ExecutionException.class, () -> result.get(3, TimeUnit.SECONDS)); else assertTrue(result.cancel(true)); @@ -370,11 +669,11 @@ void closingNonOwningFacadePreventsNewWorkButDoesNotCloseSharedTransport() throw closedClient.close(); assertThrows( CompletionException.class, - () -> closedClient.sendMessage("test", MessageContent.text("no")).join()); + () -> closedClient.sendTask("test", MessageContent.text("no")).join()); assertEquals( "TASK_STATE_COMPLETED", new DefaultWorkflowEngineClient(transport) - .sendMessage("test", MessageContent.text("yes")) + .sendTask("test", MessageContent.text("yes")) .join() .getTaskState()); } 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 f99ded30..96d4ffaf 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 @@ -80,77 +80,6 @@ void sendsFinalAuthorizationWithoutGeneratingOrSharingContext() throws Exception } } - @Test - void sendsOneShotTaskWithFreshContextAndNoControlPoint() throws Exception { - List contexts = new CopyOnWriteArrayList<>(); - List taskTexts = new CopyOnWriteArrayList<>(); - try (var transport = - new A2ATransport( - List.of(card()), - DefaultWorkflowEngineClientNegotiationTest.runtime( - params -> { - contexts.add(params.message().contextId()); - params - .message() - .parts() - .forEach( - part -> { - if (part instanceof TextPart textPart) taskTexts.add(textPart.text()); - }); - return List.of( - DefaultWorkflowEngineClientNegotiationTest.response( - params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("task-done"))); - }), - WorkflowEngineClientConfig.builder().build())) { - var sender = new DefaultExtensionSender(transport); - - var result1 = sender.sendTask("test", MessageContent.text("diagnose port")).join(); - sender.sendTask("test", MessageContent.text("diagnose port")).join(); - - assertEquals("TASK_STATE_COMPLETED", result1.getTaskState()); - assertNotEquals(contexts.get(0), contexts.get(1)); - assertEquals(2, taskTexts.size()); - assertEquals("diagnose port", taskTexts.get(0)); - assertEquals("diagnose port", taskTexts.get(1)); - } - } - - @Test - void oneShotTaskReturnsInputRequiredResultWithoutNegotiation() throws Exception { - try (var transport = - new A2ATransport( - List.of(card()), - DefaultWorkflowEngineClientNegotiationTest.runtime( - params -> - List.of( - DefaultWorkflowEngineClientNegotiationTest.response( - params, - TaskState.TASK_STATE_INPUT_REQUIRED, - MessageContent.text("need-more-info")))), - WorkflowEngineClientConfig.builder().build())) { - var result = - new DefaultExtensionSender(transport) - .sendTask("test", MessageContent.text("diagnose")) - .join(); - assertEquals("TASK_STATE_INPUT_REQUIRED", result.getTaskState()); - } - } - - @Test - void oneShotTaskFailsFastForUnknownAgent() throws Exception { - try (var transport = - new A2ATransport( - List.of(card()), - DefaultWorkflowEngineClientNegotiationTest.runtime(params -> List.of()), - WorkflowEngineClientConfig.builder().build())) { - var sender = new DefaultExtensionSender(transport); - assertThrows( - IllegalArgumentException.class, () -> sender.sendTask("unknown", MessageContent.text("x"))); - assertThrows( - NullPointerException.class, () -> sender.sendTask("test", null)); - } - } - @Test void earlyEventCanCloseHandleAndCompletionWaitsForActualExit() throws Exception { CountDownLatch callbackRan = new CountDownLatch(1); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java index 78ac0e60..197bfdfc 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java @@ -53,10 +53,12 @@ void waitsForTheOtherAgentAndSkipsMergeWhileReportingCorrelatedFailure(int statu var agent2 = new CompletableFuture(); var observed = new java.util.concurrent.CopyOnWriteArrayList>(); var client = new StubWorkflowEngineClient() { - @Override public CompletableFuture dispatch( - TaskRequest request, MessageContent content, ControlPoint callbacks) { + @Override public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { started.countDown(); - return request.getAgentName().equals("agent1") ? agent1 : agent2; + return agentName.equals("agent1") ? agent1 : agent2; } }; var merges = new AtomicInteger(); @@ -129,8 +131,10 @@ void protocolFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs( var negotiations = new AtomicInteger(); var client = new StubWorkflowEngineClient("remote-agent") { @Override - public CompletableFuture dispatch( - TaskRequest request, MessageContent content, ControlPoint callbacks) { + public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { calls.incrementAndGet(); return CompletableFuture.failedFuture(new CompletionException(remoteError)); } 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 9bf2e162..5aa3916b 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 @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -1364,4 +1365,71 @@ public long callbackTimeoutSeconds() { assertEquals(0, stub.getSentCount()); } } + + @Test + void workflowNegotiationKeepsTheOriginalTaskRequestBehindThePublicTaskApi() { + java.util.concurrent.atomic.AtomicReference prepared = + new java.util.concurrent.atomic.AtomicReference<>(); + java.util.concurrent.atomic.AtomicReference negotiated = + new java.util.concurrent.atomic.AtomicReference<>(); + StubWorkflowEngineClient client = + new StubWorkflowEngineClient("A") { + @Override + public CompletableFuture sendTask( + String agentName, + MessageContent content, + dev.openan.workflow.engine.control.NegotiationStrategy negotiationStrategy) { + TaskRequest synthetic = + TaskRequest.builder() + .agentName(agentName) + .executionId("standalone") + .taskId("standalone") + .build(); + var negotiation = + new dev.openan.workflow.engine.model.NegotiationRequest( + synthetic, + content, + new dev.openan.workflow.engine.model.ReceivedMessage( + MessageContent.text("need input"), Map.of(), List.of()), + List.of(), + java.time.Duration.ofSeconds(1)); + return negotiationStrategy + .resolve(negotiation) + .thenCompose(ignored -> sendTask(agentName, content)); + } + }; + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> { + prepared.set(request); + return CompletableFuture.completedFuture(MessageContent.text("execute")); + }) + .onNegotiation( + request -> { + negotiated.set(request.task()); + return CompletableFuture.completedFuture( + new dev.openan.workflow.engine.model.NegotiationReply.Stop( + "test-stop", "test completed")); + }) + .build(); + Workflow workflow = + Workflow.builder() + .name("negotiation-context") + .steps( + List.of( + WorkflowStep.builder() + .name("remote") + .subtasks(List.of(task("A", "work"))) + .build())) + .build(); + + ExecutionResult result = + new WorkflowExecutor(workflow, callbacks, client, null, "intent", "zh").run().join(); + + assertTrue(result.isSuccess()); + assertSame(prepared.get(), negotiated.get()); + assertEquals("remote", negotiated.get().getStepName()); + assertEquals("work", negotiated.get().getInstruction()); + } } From b82aeb77e4d14a28fd93eed61337eb1a98ffc665 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Sat, 12 Sep 2026 10:36:19 +0800 Subject: [PATCH 3/3] docs(client): document standalone task execution Signed-off-by: Zhoujie628 --- CHANGELOG.md | 10 +++++----- docs/en/API_REFERENCE.md | 28 +++++++++++++++++----------- docs/en/BUSINESS_CALLBACKS.md | 12 ++++++++++-- docs/en/DESIGN.md | 8 +++++--- docs/en/DEVELOPER_GUIDE.md | 3 ++- docs/en/INTEGRATION_GUIDE.md | 29 ++++++++++++++++++++++------- docs/zh/API_REFERENCE.md | 19 +++++++++++++------ docs/zh/BUSINESS_CALLBACKS.md | 11 ++++++++--- docs/zh/DESIGN.md | 9 +++++---- docs/zh/DEVELOPER_GUIDE.md | 4 ++-- docs/zh/INTEGRATION_GUIDE.md | 24 +++++++++++++++++++----- 11 files changed, 108 insertions(+), 49 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4346537..bb76b381 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,11 +2,11 @@ ## [Unreleased] -- Add `ExtensionSender.sendTask(agentName, content)`: a one-shot task dispatch independent of any - workflow. The host supplies final content; the engine envelopes, authenticates, sends and returns - the complete `SendMessageResult`. No `ControlPoint` is invoked and no negotiation is attempted -- - `INPUT_REQUIRED` is returned to the caller as-is. Each call uses a fresh context, matching - `sendAuthorization`. +- Add `WorkflowEngineClient.sendTask(...)` for task execution outside a DAG. It reuses the normal + task state machine, keeps one task/context through Negotiation-T, and supports a per-call + `NegotiationStrategy`. Local interaction failures cancel a known non-final remote task before + releasing the conversation. The internal workflow dispatch operation and the former + `sendMessage(...)` alias are no longer part of the public `WorkflowEngineClient` contract. - Increase the default Notification-T acknowledgement timeout from 5 seconds to 5 minutes and expose the sample Spring setting as `a2a.notification-ack-timeout-seconds` / `A2A_NOTIFICATION_ACK_TIMEOUT_SECONDS`. diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index e30708b6..5ab6e536 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -58,8 +58,9 @@ ExecutionResult result = ExecutePsop.builder() ### WorkflowEngineClient ```java -CompletableFuture dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks); -CompletableFuture sendMessage(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy); CompletableFuture getTask(String agentName, String taskId); CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); @@ -75,7 +76,18 @@ AgentCard, authentication and custom runtime as message dispatch. A list result 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 executor calls dispatch; onTask does not send. Content is final parts/metadata/extensions. The engine manages +`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 +a per-call strategy without changing the client-wide ControlPoint. If a known remote task remains +non-final after a timeout, missing handler or other local interaction failure, the client attempts +to cancel it before completing exceptionally. `sendTask` is the only public task-submission name; +protocol-runtime implementations may still use the A2A operation name `sendMessage` internally. + +Plain A2A content does not require Task-T. When Task-T is activated, its metadata must be present and +the target AgentCard must declare Task-T support. + +After `onTask` returns, the executor calls `sendTask` internally; `onTask` does not send. Content is final +parts/metadata/extensions. The engine manages envelopes and interactions, never instantiates A2ATClient or generates content from AgentCard declarations. Template queries and generation belong to the host SDK. @@ -86,7 +98,8 @@ obtain the received context; reply with the same id, round and maxRounds. The la Do not call nextRound for an ending reply or return a new Propose. Return `new NegotiationReply.Send(content)` to send that exact content. Return `new NegotiationReply.Stop(code, reason)` -to stop locally without a generated Abort. Repeated task/session/round events do not repeat the callback or submission. +to stop without generating Abort content; the engine then cancels the known non-final A2A task as lifecycle cleanup. +Repeated task/session/round events do not repeat the callback or submission. Unchanged waiting state is observed with getTask. `maxNegotiationExchanges` (default 3) bounds local interactions, independently of the SDK context's maxRounds. Timeout, exhausted budget or a missing handler fails locally; no implicit Accept or synthesized Abort. Accept/Reject ACKs in @@ -97,17 +110,10 @@ task success, even if the dispatched agent acknowledges it with COMPLETED. ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); -CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` -`sendTask` dispatches a one-shot task without a workflow: the host supplies final content (structured -Task-T prompt or plain text), the engine envelopes, authenticates and sends it, and returns the -complete `SendMessageResult`. No `ControlPoint` is invoked and no negotiation is attempted -- if the -agent responds with `INPUT_REQUIRED`, the result carries that state and the caller decides the next -step. Each call uses a fresh context independent of any workflow session. - Authorization and notification accept host-generated final content. Use separate transport/runtime/context instances; 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; diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index d98bac23..6406e4b0 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -132,7 +132,8 @@ obtain the received context; reply with the same id, round and maxRounds. The la Do not call nextRound for an ending reply or return a new Propose. Return `new NegotiationReply.Send(content)` to send that exact content. Return `new NegotiationReply.Stop(code, reason)` -to stop locally without a generated Abort. Repeated task/session/round events do not repeat the callback or submission. +to stop without generating Abort content; the engine then cancels the known non-final A2A task as lifecycle cleanup. +Repeated task/session/round events do not repeat the callback or submission. Unchanged waiting state is observed with getTask. `maxNegotiationExchanges` (default 3) bounds local interactions, independently of the SDK context's maxRounds. Timeout, exhausted budget or a missing handler fails locally; no implicit Accept or synthesized Abort. Accept/Reject ACKs in @@ -204,11 +205,18 @@ output comes as JSON in outputs; natural-language output calls for text matching Callbacks for different conditional edges may run concurrently; do not hold shared mutable current-task state. Each workflow task activation (preparation and dispatch combined) is bounded by the client timeout (default sendTimeoutSeconds=600). Routing has a -callback timeout; dispatch/negotiation additionally has its own total wait deadline. Cancellation/timeout prevents late +callback timeout; task interaction/negotiation additionally has its own total wait deadline. Cancellation/timeout prevents late sends; it does not automatically stop external business/LLM work. Hosts own cancellation of their resources. Synchronous callback entry points must return promptly; place blocking operations in an asynchronous executor. Missing/null/exceptional callbacks fail; uncertain sends are not blindly retried. +### 6.1 Standalone task + +Use `WorkflowEngineClient.sendTask(agentName, finalContent)` for a task outside the DAG. The overload +with `NegotiationStrategy` supplies only that call's negotiation decisions. The host still owns +content generation and semantic validation; the engine preserves the remote task/context, waits for +a final state and cleans up a known non-final task when local interaction cannot continue. + ## 7. Independent extensions ExtensionSender.sendAuthorization(agentName, finalContent) returns CompletableFuture. diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index 5c94b729..966acb39 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -119,7 +119,7 @@ entry point. graph TD L2["Layer 2 - Orchestration
    execute_psop / ExecutePsop
    lifecycle, event stream, cancellation, onFinish persistence"] L1["Layer 1 - Traversal
    WorkflowExecutor
    DAG walk, parallel dispatch, context assembly, routing"] - L0["Layer 0 - Communication
    A2ATransport + two facades
    WorkflowEngineClient (workflow send) | ExtensionSender (independent operations)"] + L0["Layer 0 - Communication
    A2ATransport + two facades
    WorkflowEngineClient (workflow / standalone tasks) | ExtensionSender (independent operations)"] F["Foundation - Decision
    ControlPoint
    user-implemented business decisions"] L2 --> L1 --> L0 @@ -204,7 +204,8 @@ callback budget expires or the run is cancelled is ignored and never sent to the ### 3.1 Layer 0 - Communication A2ATransport uses A2A SDK REST, JSON-RPC and gRPC bindings for authentication, delivery and complete response assembly. -WorkflowEngineClient accepts MessageContent and manages task association and negotiation continuation. ExtensionSender +WorkflowEngineClient accepts MessageContent for workflow or standalone tasks and manages task association, waiting, +negotiation continuation and non-final task cleanup. ExtensionSender provides independent sendAuthorization and openNotification operations. The implementations are reusable; task, authorization and notification do not share transport/runtime/context instances. ProtocolResponses assembles artifact deltas by identity and ReceivedMessage preserves metadata at each level. @@ -267,7 +268,8 @@ obtain the received context; reply with the same id, round and maxRounds. The la Do not call nextRound for an ending reply or return a new Propose. Return `new NegotiationReply.Send(content)` to send that exact content. Return `new NegotiationReply.Stop(code, reason)` -to stop locally without a generated Abort. Repeated task/session/round events do not repeat the callback or submission. +to stop without generating Abort content; the engine then cancels the known non-final A2A task as lifecycle cleanup. +Repeated task/session/round events do not repeat the callback or submission. Unchanged waiting state is observed with getTask. `maxNegotiationExchanges` (default 3) bounds local interactions, independently of the SDK context's maxRounds. Timeout, exhausted budget or a missing handler fails locally; no implicit Accept or synthesized Abort. Accept/Reject ACKs in diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 18aa32a4..2e8da7b9 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -136,7 +136,8 @@ obtain the received context; reply with the same id, round and maxRounds. The la Do not call nextRound for an ending reply or return a new Propose. Return `new NegotiationReply.Send(content)` to send that exact content. Return `new NegotiationReply.Stop(code, reason)` -to stop locally without a generated Abort. Repeated task/session/round events do not repeat the callback or submission. +to stop without generating Abort content; the engine then cancels the known non-final A2A task as lifecycle cleanup. +Repeated task/session/round events do not repeat the callback or submission. Unchanged waiting state is observed with getTask. `maxNegotiationExchanges` (default 3) bounds local interactions, independently of the SDK context's maxRounds. Timeout, exhausted budget or a missing handler fails locally; no implicit Accept or synthesized Abort. Accept/Reject ACKs in diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 2d8fa8b9..d1d5f9f3 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -393,24 +393,39 @@ obtain the received context; reply with the same id, round and maxRounds. The la Do not call nextRound for an ending reply or return a new Propose. Return `new NegotiationReply.Send(content)` to send that exact content. Return `new NegotiationReply.Stop(code, reason)` -to stop locally without a generated Abort. Repeated task/session/round events do not repeat the callback or submission. +to stop without generating Abort content; the engine then cancels the known non-final A2A task as lifecycle cleanup. +Repeated task/session/round events do not repeat the callback or submission. Unchanged waiting state is observed with getTask. `maxNegotiationExchanges` (default 3) bounds local interactions, independently of the SDK context's maxRounds. Timeout, exhausted budget or a missing handler fails locally; no implicit Accept or synthesized Abort. Accept/Reject ACKs in SUBMITTED/WORKING remain pending and are observed without resending the command. A business-sent Abort is never task success, even if the dispatched agent acknowledges it with COMPLETED. +For a task outside a workflow DAG, call the task client directly. It uses a fresh context but keeps +the same remote task identity through waiting and negotiation. The optional per-call strategy does +not change the client-wide ControlPoint. Plain A2A content is allowed; activated Task-T content must +include Task-T metadata and the target AgentCard must declare the extension. + ```java -CompletableFuture sendAuthorization(String agentName, MessageContent content); CompletableFuture sendTask(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy); +``` + +If a known remote task is still non-final when local interaction cannot continue, the client makes +a best-effort cancellation before completing exceptionally. Standard A2A request errors remain +exceptional results and are not converted into business task failures. + +```java +CompletableFuture sendAuthorization(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` -The host agent generates final Authorization-T/Notification-T content and calls these methods on separate -transport/runtime/context instances. The listener receives the handle and 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. +The host agent generates final Authorization-T/Notification-T content. Authorization and notification each use a +transport/runtime/context independent of the task client and of one another. The listener receives the handle and +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. `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 @@ -636,7 +651,7 @@ Logging configuration, pretty display and observer callbacks do not determine ta |--------------------------------------------------------|-------------------------------------------------------------------------| | `ExecutePsop.Builder` | Workflow execution entry point | | `ControlPoint` / `DefaultControlPoint` | Business decisions (onTask, onSelfTask, onRoute, onNegotiation, etc.) | -| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | Workflow send (sendMessage, auth, extensions) | +| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | Workflow and standalone task interaction | | `ExtensionSender` / `DefaultExtensionSender` | Independent Authorization-T operations and Notification-T subscriptions | | `A2ATransport` | Shared wire layer (A2A Java client runtime, auth, SSE consumer) | | `WorkflowEngineClientConfig` | Configuration (TLS, auth, deadlines, executor limits, negotiation exchange budget) | diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index d069eef1..6d2c7909 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -58,8 +58,9 @@ ExecutionResult result = ExecutePsop.builder() ### WorkflowEngineClient ```java -CompletableFuture dispatch(TaskRequest request, MessageContent content, ControlPoint callbacks); -CompletableFuture sendMessage(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy); CompletableFuture getTask(String agentName, String taskId); CompletableFuture listTasks(String agentName, ListTasksParams params); CompletableFuture cancelTask(String agentName, String taskId); @@ -74,7 +75,14 @@ void close(); 任务管理操作。查询结果受认证身份权限约束，只包含当前身份可见的任务；调用方必须使用 `nextPageToken` 完成分页。关闭本地客户端或通知流不等同于取消远端任务。 -执行器内部调用 dispatch；onTask 不自行发送。内容是最终 parts/metadata/extensions，引擎只管理信封和交互，不创建 A2ATClient，不按 +`sendTask` 在 DAG 外执行宿主提供的最终任务内容。每次调用使用新的 context，并复用工作流任务的发送、等待、查询和 +Negotiation-T 闭环。重载方法接受本次调用独有的协商策略，不修改客户端级 ControlPoint。超时、缺少处理器或其他本地交互失败后， +如果已知远端任务仍处于非终态，客户端会先尝试取消，再以异常完成。`sendTask` 是公开任务提交的唯一名称；传输运行时内部仍可使用 +A2A 标准动作名称 `sendMessage`。 + +普通 A2A 内容不要求 Task-T；一旦激活 Task-T，就必须同时携带对应 metadata，且目标 AgentCard 必须声明支持 Task-T。 + +`onTask` 返回后由执行器内部调用 `sendTask`；`onTask` 不自行发送。内容是最终 parts/metadata/extensions，引擎只管理信封和交互，不创建 A2ATClient，不按 AgentCard 声明生成内容。模板查询和生成接口请直接使用宿主 SDK。 只有远端 `INPUT_REQUIRED` 携带有效 Negotiation-T Propose 才进入 `onNegotiation`。 终态不会重启协商，普通 INPUT_REQUIRED @@ -82,8 +90,8 @@ AgentCard 声明生成内容。模板查询和生成接口请直接使用宿主 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 -返回 `new NegotiationReply.Send(content)` 发送最终内容； 返回 `new NegotiationReply.Stop(code, reason)` 只在本地停止，不生成 -Abort。 同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 @@ -92,7 +100,6 @@ COMPLETED 确认，也不能判为任务成功。 ```java CompletableFuture sendAuthorization(String agentName, MessageContent content); -CompletableFuture sendTask(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 79f60453..282bf59f 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -114,8 +114,8 @@ TaskRequest，originalSubmission 是首次最终提交， received 是当前完 `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 -返回 `new NegotiationReply.Send(content)` 发送最终内容； 返回 `new NegotiationReply.Stop(code, reason)` 只在本地停止，不生成 -Abort。 同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 @@ -179,10 +179,15 @@ onRoute 每次收到一条条件边，从 `currentResults()` 读取本步骤刚 `currentResults()` 输出的真实解析（结构化输出用 outputs 里的 JSON，自然语言输出用文本包含或宿主自己的判定逻辑）。 不同条件边的回调可能并发，不要共享可变的“当前任务”状态。每个工作流任务从内容准备到传输完成受客户端 timeout 总体限制，默认 -sendTimeoutSeconds=600； 路由单独限制回调等待时间，dispatch／协商另有总等待截止时间。取消／超时后晚到结果不发送， +sendTimeoutSeconds=600；路由单独限制回调等待时间，任务交互／协商另有总等待截止时间。取消／超时后晚到结果不发送， 但不等于自动取消宿主正在运行的 LLM 或业务操作，宿主负责清理其资源。 同步回调入口应迅速返回，阻塞任务应交给异步执行器。 回调缺失、返回 null 或异常均明确失败；不确定发送失败不自动重发。 +### 6.1 独立任务 + +DAG 外任务使用 `WorkflowEngineClient.sendTask(agentName, finalContent)`。带 `NegotiationStrategy` 的重载只提供本次调用的 +协商决策。内容生成和语义校验仍由宿主负责；引擎保持远端 task/context、等待终态，并在本地交互无法继续时清理已知的非终态任务。 + ## 7. 独立授权与订阅 ExtensionSender.sendAuthorization(agentName, finalContent) 返回 CompletableFuture。 diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index 41cbe230..aee0bb9d 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -108,7 +108,7 @@ SDK 分为四层，每层构建在下一层之上，单一职责，入口清晰 graph TD L2["Layer 2 - 编排层
    ExecutePsop
    生命周期、事件流、取消、onFinish 持久化"] L1["Layer 1 - 遍历层
    WorkflowExecutor
    DAG 遍历、并行下发、上下文组装、路由"] - L0["Layer 0 - 通信层
    A2ATransport + 两个门面
    WorkflowEngineClient（工作流发送）| ExtensionSender（独立协议操作）"] + L0["Layer 0 - 通信层
    A2ATransport + 两个门面
    WorkflowEngineClient（工作流／独立任务）| ExtensionSender（独立协议操作）"] F["基础层 - 决策
    ControlPoint
    用户实现的业务决策"] L2 --> L1 --> L0 @@ -188,7 +188,8 @@ A2A 错误信封拒绝的调用，由传输层识别并投影为稳定错误码 ### 3.1 Layer 0 - 通信层 A2ATransport 使用 A2A SDK 的 REST、JSON-RPC、gRPC 绑定，负责认证头、实际传输和完整响应组装。 WorkflowEngineClient 接收 -MessageContent 并管理远端任务和必要的协商续发。 ExtensionSender 负责流程外 sendAuthorization 与 openNotification。 +工作流或独立任务的 MessageContent，并管理远端任务、等待、必要的协商续发与非终态清理。 ExtensionSender 负责流程外 +sendAuthorization 与 openNotification。 三种协议操作复用实现而不共用 transport/runtime/context 实例。 ProtocolResponses 按 artifact 身份合并流式增量，ReceivedMessage 保留各层 metadata。 @@ -240,8 +241,8 @@ Negotiation-T： `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 -返回 `new NegotiationReply.Send(content)` 发送最终内容； 返回 `new NegotiationReply.Stop(code, reason)` 只在本地停止，不生成 -Abort。 同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index e3834b36..04092ab1 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -130,8 +130,8 @@ try (var client = new DefaultWorkflowEngineClient(agentCards, a2aRuntime, `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 -返回 `new NegotiationReply.Send(content)` 发送最终内容； 返回 `new NegotiationReply.Stop(code, reason)` 只在本地停止，不生成 -Abort。 同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 72bc0ccb..aaa0dcc4 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -376,20 +376,34 @@ AgentCard 通过 `capabilities.extensions` 声明扩展点： `A2atMessages.contextOf(request.received())` 取得收到的上下文； 结束回复保持相同 id、round、maxRounds，最后允许的一轮仍可回答，不自行 nextRound 或返回新 Propose。 -返回 `new NegotiationReply.Send(content)` 发送最终内容； 返回 `new NegotiationReply.Stop(code, reason)` 只在本地停止，不生成 -Abort。 同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 +返回 `new NegotiationReply.Send(content)` 发送最终内容；返回 `new NegotiationReply.Stop(code, reason)` 时不生成 Abort 内容， +引擎随后通过取消已知的非终态 A2A 任务完成生命周期清理。同一任务／会话／轮次的重复等待事件不会重复回调、重复提交；未变化状态通过 getTask 观察。 `maxNegotiationExchanges` 默认 3，是独立于 SDK context.maxRounds 的本地交互资源预算。 超时、预算耗尽、回调缺失均明确失败，不默认 Accept，也不自动生成 Abort。 Accept/Reject 的 SUBMITTED/WORKING ACK 仍需等待任务结果，不重发原命令。 业务发送 Abort 后，即使远端用 COMPLETED 确认，也不能判为任务成功。 +对于 DAG 外的独立任务，直接调用任务客户端。每次调用使用新的 context，但等待和协商续发始终保持同一个远端 taskId。 +可选的单次调用策略不会修改客户端级 ControlPoint。普通 A2A 内容可以直接发送；激活 Task-T 时必须携带 Task-T metadata， +且目标 AgentCard 必须声明该扩展。 + ```java -CompletableFuture sendAuthorization(String agentName, MessageContent content); CompletableFuture sendTask(String agentName, MessageContent content); +CompletableFuture sendTask(String agentName, MessageContent content, + NegotiationStrategy negotiationStrategy); +``` + +本地交互无法继续且已知远端任务仍为非终态时，客户端先尽力取消远端任务，再以异常完成。标准 A2A 请求错误仍按异常传递， +不会伪装成业务任务失败。 + +```java +CompletableFuture sendAuthorization(String agentName, MessageContent content); NotificationSubscription openNotification(String agentName, MessageContent content, BiConsumer listener); ``` -宿主智能体生成最终 Authorization-T/Notification-T 内容后调用上述接口；使用三类独立 transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage，在宿主定义的终态事件上关闭。handle.acknowledgement() 和 completion() 分别表示 ACK 和真实流退出，两者都不是工作流前提。 +宿主智能体生成最终 Authorization-T/Notification-T 内容。授权与通知分别使用独立于任务客户端且彼此隔离的 +transport/runtime/context。订阅监听器收到 handle 与完整 ReceivedMessage，在宿主定义的终态事件上关闭。 +handle.acknowledgement() 和 completion() 分别表示 ACK 和真实流退出，两者都不是工作流前提。 `WorkflowEngineClientConfig.notificationAckTimeoutSeconds` 控制首次订阅 ACK 的等待时间，默认 300 秒；Spring 示例可通过 `a2a.notification-ack-timeout-seconds` 或环境变量 `A2A_NOTIFICATION_ACK_TIMEOUT_SECONDS` 覆盖。该参数不控制 ACK 后的 SSE @@ -592,7 +606,7 @@ contextId 与 executionId，供跨层定位。已识别的 A2A 错误使用 WARN |--------------------------------------------------------|---------------------------------------------------------------| | `ExecutePsop.Builder` | 工作流执行入口 | | `ControlPoint` / `DefaultControlPoint` | 业务决策实现（onTask、onSelfTask、onRoute、onNegotiation 等） | -| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | 工作流发送（sendMessage、认证、扩展） | +| `WorkflowEngineClient` / `DefaultWorkflowEngineClient` | 工作流与独立任务交互 | | `ExtensionSender` / `DefaultExtensionSender` | 独立 Authorization-T 操作与 Notification-T 长连接订阅 | | `A2ATransport` | 共享通信层（A2A Java 客户端 runtime、认证、SSE 消费） | | `WorkflowEngineClientConfig` | 配置（SSL、认证、A2A-T、协商轮数、自定义 Handler） |