From a134c2c190d1d7df8b785e01f8861da9eb47f99e Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Fri, 4 Sep 2026 20:35:37 +0800 Subject: [PATCH 1/4] feat(engine): adopt the standard A2A error envelope Detect the google.rpc.Status error envelope on direct responses and SSE data, projecting httpStatus, code, status, reason, domain, typed details and Retry-After into stable workflow error codes (a2a., with an a2a.http. fallback when no reason is observed). Remove the legacy {status,title,detail} problem compatibility (RemoteProblemException, ProblemDetectingHttpClient) per the agreement that OMC returns the standard envelope for pre-task-creation failures. Fail immediately when an SSE stream completes without delivering any event instead of waiting out the send timeout; the observed HTTP status is not fabricated when the transport cannot expose it. Stop logging subscriber-initiated cancellations as transport failures and redact failure messages in protocol logs. Signed-off-by: Zhoujie628 --- ....java => A2AErrorDetectingHttpClient.java} | 33 ++- .../workflow/engine/client/A2ATransport.java | 12 +- .../client/DefaultA2AJavaClientRuntime.java | 18 +- .../engine/client/ObservedHttpClient.java | 11 +- .../client/RemoteA2AErrorException.java | 243 ++++++++++++++++++ .../engine/client/RemoteProblemException.java | 92 ------- .../workflow/engine/core/FailureMapping.java | 31 ++- ...a => A2AErrorDetectingHttpClientTest.java} | 56 ++-- .../engine/client/ObservedHttpClientTest.java | 114 ++++++++ .../client/RemoteA2AErrorResponseTest.java | 194 ++++++++++++++ .../client/RemoteProblemResponseTest.java | 152 ----------- ...st.java => RemoteA2AErrorFailureTest.java} | 102 ++++---- 12 files changed, 711 insertions(+), 347 deletions(-) rename workflow-engine/src/main/java/dev/openan/workflow/engine/client/{ProblemDetectingHttpClient.java => A2AErrorDetectingHttpClient.java} (81%) create mode 100644 workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java delete mode 100644 workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java rename workflow-engine/src/test/java/dev/openan/workflow/engine/client/{ProblemDetectingHttpClientTest.java => A2AErrorDetectingHttpClientTest.java} (67%) create mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java delete mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java rename workflow-engine/src/test/java/dev/openan/workflow/engine/core/{RemoteProblemFailureTest.java => RemoteA2AErrorFailureTest.java} (61%) diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java similarity index 81% rename from workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java rename to workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java index e4da1a1d..c602404e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java @@ -5,16 +5,16 @@ * 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 + * 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. + * 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; @@ -28,11 +28,11 @@ import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.client.http.ServerSentEvent; -/** Checks problem responses before the SDK's A2A parser; wire logging remains observational. */ -final class ProblemDetectingHttpClient implements A2AHttpClient { +/** Detects standard A2A error envelopes before the SDK attempts to parse them as task events. */ +final class A2AErrorDetectingHttpClient implements A2AHttpClient { private final A2AHttpClient delegate; - ProblemDetectingHttpClient(A2AHttpClient delegate) { + A2AErrorDetectingHttpClient(A2AHttpClient delegate) { this.delegate = delegate; } @@ -41,7 +41,8 @@ final class ProblemDetectingHttpClient implements A2AHttpClient { @Override public DeleteBuilder createDelete() { return new Delete(delegate.createDelete()); } private static A2AHttpResponse check(A2AHttpResponse response) { - var error = RemoteProblemException.fromPayload(response.body()); + var error = RemoteA2AErrorException.fromResponse( + response.status(), response.body(), response.headers().toMap()); if (error != null) throw error; return response; } @@ -90,6 +91,7 @@ private static CompletableFuture stream( var result = new CompletableFuture(); var upstream = new AtomicReference>(); var ended = new AtomicBoolean(); + var eventReceived = new AtomicBoolean(); Consumer fail = error -> { if (!ended.compareAndSet(false, true)) return; try { errors.accept(error); } @@ -107,13 +109,18 @@ private static CompletableFuture stream( }); var active = starter.start(event -> { if (ended.get()) return; - var problem = RemoteProblemException.fromPayload(event.data()); - if (problem != null) fail.accept(problem); + var remoteError = RemoteA2AErrorException.fromPayload(event.data()); + if (remoteError != null) fail.accept(remoteError); else { + eventReceived.set(true); try { messages.accept(event); } catch (RuntimeException callbackError) { fail.accept(callbackError); } } }, fail, () -> { + if (!eventReceived.get()) { + fail.accept(new IOException("A2A streaming response closed without an event")); + return; + } if (ended.compareAndSet(false, true)) { try { complete.run(); result.complete(null); } catch (RuntimeException error) { result.completeExceptionally(error); } 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 b3b9ca06..3b87d049 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 @@ -488,14 +488,16 @@ CompletableFuture send( .receivedMessages(ProtocolResponses.assemble(events)) .build(); } catch (Exception e) { - RemoteProblemException problem = RemoteProblemException.findIn(e); - if (problem != null) { + RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(e); + if (remoteError != null) { log.warn( - "[Transport] REMOTE_PROBLEM agent={}, contextId={}, status={}, reason={}", + "[Transport] A2A_ERROR agent={}, contextId={}, httpStatus={}, status={}, reason={}, message={}", agentName, contextId, - problem.getStatus(), - problem.getMessage().replace("\r", "\\r").replace("\n", "\\n")); + remoteError.getHttpStatus(), + remoteError.getStatus(), + remoteError.getReason(), + remoteError.getMessage().replace("\r", "\\r").replace("\n", "\\n")); } else { log.error( "[Transport] Failed to send message to {}: {}", 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 4abc929e..14ca7bcb 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 @@ -216,18 +216,21 @@ private static void onError( error.getMessage()); } else { String msg = error.getMessage() != null ? error.getMessage() : ""; - RemoteProblemException problem = RemoteProblemException.findIn(error); - boolean connectionClosed = problem == null && TransportFailures.isExpectedLocalClose(error); + RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(error); + boolean connectionClosed = + remoteError == null && TransportFailures.isExpectedLocalClose(error); if (connectionClosed) { log.debug("[A2ARuntime] Connection closed for '{}': {}", agentName, msg); } else { errorRef.set(error); - if (problem != null) { + if (remoteError != null) { log.warn( - "[A2ARuntime] REMOTE_PROBLEM agent={}, status={}, reason={}", + "[A2ARuntime] A2A_ERROR agent={}, httpStatus={}, status={}, reason={}, message={}", agentName, - problem.getStatus(), - problem.getMessage().replace("\r", "\\r").replace("\n", "\\n")); + remoteError.getHttpStatus(), + remoteError.getStatus(), + remoteError.getReason(), + remoteError.getMessage().replace("\r", "\\r").replace("\n", "\\n")); } else { log.error( "[A2ARuntime] Error callback for '{}': {}", agentName, error.getMessage(), error); @@ -695,7 +698,8 @@ private A2AHttpClient createHttpClient() { crlPath, Duration.ofSeconds(60), httpClientExecutor); - return new ProblemDetectingHttpClient(new JdkA2AHttpClient(new ObservedHttpClient(httpClient))); + return new A2AErrorDetectingHttpClient( + new JdkA2AHttpClient(new ObservedHttpClient(httpClient))); } private record StreamClientKey(String agentName, String contextId) {} 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 3a42077a..ab78645e 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 @@ -19,6 +19,7 @@ package dev.openan.workflow.engine.client; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; import java.io.IOException; import java.net.*; import java.net.http.*; @@ -145,6 +146,8 @@ private final class Exchange { final Map context = WireLog.context(); final java.util.concurrent.atomic.AtomicBoolean failureLogged = new java.util.concurrent.atomic.AtomicBoolean(); + final java.util.concurrent.atomic.AtomicBoolean cancelledBySubscriber = + new java.util.concurrent.atomic.AtomicBoolean(); Exchange(HttpRequest request) { this.request = request; @@ -174,15 +177,20 @@ void emit( } void failure(Throwable error) { + if (cancelledBySubscriber.get()) return; // Body subscriber and sendAsync completion can report the same exchange failure. if (!failureLogged.compareAndSet(false, true)) return; + String message = SensitiveDataRedactor.redact(Objects.toString(error.getMessage(), "")) + .replace("\r", "\\r") + .replace("\n", "\\n"); emit( "FAILURE", null, Map.of(), "transport-error", "", - "errorType=" + error.getClass().getSimpleName()); + "errorType=" + error.getClass().getSimpleName() + + (message.isBlank() ? "" : "; errorMessage=" + message)); } HttpRequest observedRequest() { @@ -312,6 +320,7 @@ public void request(long count) { @Override public void cancel() { + cancelledBySubscriber.set(true); body.end(true); subscription.cancel(); } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java new file mode 100644 index 00000000..b820f268 --- /dev/null +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteA2AErrorException.java @@ -0,0 +1,243 @@ +/* + * 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 com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; +import java.util.ArrayList; +import java.util.Collections; +import java.util.IdentityHashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import org.a2aproject.sdk.spec.A2AClientHTTPError; +import org.a2aproject.sdk.spec.A2AError; +import org.a2aproject.sdk.spec.A2AErrorCodes; + +/** A standard A2A HTTP error projected from the {@code google.rpc.Status} JSON envelope. */ +public final class RemoteA2AErrorException extends RuntimeException { + private static final ObjectMapper JSON = + new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); + private static final TypeReference>> DETAIL_TYPE = + new TypeReference<>() {}; + + private final int httpStatus; + private final int code; + private final String status; + private final String reason; + private final String domain; + private final List> details; + private final Map> responseHeaders; + + private RemoteA2AErrorException( + int httpStatus, + int code, + String status, + String message, + String reason, + String domain, + List> details, + Map> responseHeaders) { + super(message); + this.httpStatus = httpStatus; + this.code = code; + this.status = status; + this.reason = reason; + this.domain = domain; + this.details = List.copyOf(details); + this.responseHeaders = SensitiveDataRedactor.safeHeaders(responseHeaders); + } + + /** Parses a top-level standard A2A error envelope whose embedded code supplies HTTP status. */ + public static RemoteA2AErrorException fromPayload(String payload) { + return fromResponse(0, payload, Map.of()); + } + + /** Parses a standard A2A error response, preferring an observed non-2xx HTTP status. */ + public static RemoteA2AErrorException fromResponse( + int observedHttpStatus, String payload, Map> responseHeaders) { + if (payload == null || payload.isBlank()) return null; + JsonNode root; + try { + root = JSON.readTree(payload); + } catch (java.io.IOException ignored) { + return null; + } + if (root == null || !root.isObject() || !root.path("error").isObject()) return null; + JsonNode error = root.path("error"); + if (!error.path("code").isIntegralNumber() || !error.path("code").canConvertToInt()) { + return null; + } + int code = error.path("code").asInt(); + int httpStatus = isErrorStatus(observedHttpStatus) ? observedHttpStatus : code; + if (!isErrorStatus(code) || !isErrorStatus(httpStatus)) return null; + String message = safeText(error, "message"); + if (message.isBlank()) return null; + String status = safeText(error, "status"); + List> details = safeDetails(error.path("details")); + String reason = ""; + String domain = ""; + if (error.path("details").isArray()) { + for (JsonNode detail : error.path("details")) { + if (!detail.isObject() || !detail.path("reason").isTextual()) continue; + reason = safeText(detail, "reason"); + domain = safeText(detail, "domain"); + break; + } + } + return new RemoteA2AErrorException( + httpStatus, + code, + status, + message, + reason, + domain, + details, + responseHeaders == null ? Map.of() : responseHeaders); + } + + /** Finds or projects a standard A2A error in a wrapped SDK/transport exception chain. */ + public static RemoteA2AErrorException findIn(Throwable error) { + Set seen = Collections.newSetFromMap(new IdentityHashMap<>()); + for (Throwable current = error; current != null && seen.add(current); current = current.getCause()) { + if (current instanceof RemoteA2AErrorException remote) return remote; + if (current instanceof A2AClientHTTPError http) { + RemoteA2AErrorException parsed = + fromResponse(http.getCode(), http.getResponseBody(), http.getResponseHeaders()); + if (parsed != null) return parsed; + return transportOnly(http); + } + if (current instanceof A2AError a2a) return fromSdkError(a2a); + } + return null; + } + + private static RemoteA2AErrorException transportOnly(A2AClientHTTPError error) { + String message = SensitiveDataRedactor.redact(error.getMessage()); + return new RemoteA2AErrorException( + error.getCode(), + error.getCode(), + "", + message.isBlank() ? "A2A request failed with HTTP " + error.getCode() : message, + "", + "", + List.of(), + error.getResponseHeaders()); + } + + private static RemoteA2AErrorException fromSdkError(A2AError error) { + A2AErrorCodes errorCode = A2AErrorCodes.fromCode(error.getCode()); + int httpStatus = errorCode == null ? 500 : errorCode.httpCode(); + String reason = errorCode == null ? "" : errorCode.name(); + String status = errorCode == null ? "INTERNAL" : errorCode.grpcStatus(); + Map detail = new LinkedHashMap<>(); + detail.put("@type", "type.googleapis.com/google.rpc.ErrorInfo"); + if (!reason.isBlank()) detail.put("reason", reason); + detail.put("domain", "a2a-protocol.org"); + if (!error.getDetails().isEmpty()) detail.put("metadata", error.getDetails()); + List> details = safeDetails(JSON.valueToTree(List.of(detail))); + return new RemoteA2AErrorException( + httpStatus, + httpStatus, + status, + SensitiveDataRedactor.redact(error.getMessage()), + reason, + "a2a-protocol.org", + details, + Map.of()); + } + + private static List> safeDetails(JsonNode node) { + if (!node.isArray()) return List.of(); + try { + String redacted = SensitiveDataRedactor.redact(JSON.writeValueAsString(node)); + List> converted = JSON.readValue(redacted, DETAIL_TYPE); + List> result = new ArrayList<>(); + for (Map detail : converted) { + result.add(Collections.unmodifiableMap(new LinkedHashMap<>(detail))); + } + return List.copyOf(result); + } catch (java.io.IOException ignored) { + return List.of(); + } + } + + private static boolean isErrorStatus(int value) { + return value >= 400 && value <= 599; + } + + private static String safeText(JsonNode node, String key) { + return node.path(key).isTextual() + ? SensitiveDataRedactor.redact(node.path(key).textValue()) + : ""; + } + + /** Stable workflow-facing error code that does not expose SDK exception class names. */ + public String workflowErrorCode() { + String normalizedReason = + reason + .toLowerCase(Locale.ROOT) + .replaceAll("[^a-z0-9]+", "_") + .replaceAll("^_+|_+$", ""); + if (!normalizedReason.isBlank()) return "a2a." + normalizedReason; + return "a2a.http." + httpStatus; + } + + public int getHttpStatus() { + return httpStatus; + } + + public int getCode() { + return code; + } + + public String getStatus() { + return status; + } + + public String getReason() { + return reason; + } + + public String getDomain() { + return domain; + } + + public List> getDetails() { + return details; + } + + public Map> getResponseHeaders() { + return responseHeaders; + } + + public String getRetryAfter() { + for (Map.Entry> header : responseHeaders.entrySet()) { + if ("retry-after".equalsIgnoreCase(header.getKey()) && !header.getValue().isEmpty()) { + return header.getValue().get(0); + } + } + return ""; + } +} diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java deleted file mode 100644 index c5e10e11..00000000 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/RemoteProblemException.java +++ /dev/null @@ -1,92 +0,0 @@ -/* - * 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 com.fasterxml.jackson.databind.DeserializationFeature; -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; - -/** - * A remote problem response, including errors sent as SSE data after HTTP 200 headers. - * Known credential fields are redacted using the same rules as protocol logging. - */ -public final class RemoteProblemException extends RuntimeException { - private static final ObjectMapper JSON = - new ObjectMapper().enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS); - private final int status; - private final String title; - private final String detail; - private final String type; - private final String timestamp; - - private RemoteProblemException(JsonNode problem) { - super("Remote agent error (status=" + problem.path("status").asInt() + "): " - + text(problem, "title") - + (text(problem, "title").isEmpty() ? "" : " — ") + text(problem, "detail")); - status = problem.path("status").asInt(); - title = text(problem, "title"); - detail = text(problem, "detail"); - type = text(problem, "type"); - timestamp = text(problem, "timestamp"); - } - - /** Returns a top-level problem error, or null for ordinary A2A events and incomplete JSON. */ - public static RemoteProblemException fromPayload(String payload) { - if (payload == null || payload.isBlank()) return null; - JsonNode node; - try { - node = JSON.readTree(payload); - } catch (java.io.IOException ignored) { - return null; - } - if (node == null || !node.isObject() || !node.path("status").isIntegralNumber() - || !node.path("status").canConvertToInt()) return null; - int status = node.path("status").asInt(); - if (status < 400 || status > 599 - || (text(node, "title").isBlank() && text(node, "detail").isBlank())) return null; - for (String envelope : new String[] {"task", "message", "statusUpdate", "artifactUpdate", "result", "error"}) { - if (node.has(envelope)) return null; - } - return new RemoteProblemException(node); - } - - private static String text(JsonNode node, String key) { - return node.path(key).isTextual() ? WireLog.redact(node.path(key).textValue()) : ""; - } - - /** Finds a problem preserved in a transport exception's cause chain, without looping on cycles. */ - public static RemoteProblemException findIn(Throwable error) { - var seen = java.util.Collections.newSetFromMap(new java.util.IdentityHashMap()); - for (Throwable current = error; current != null && seen.add(current); current = current.getCause()) { - if (current instanceof RemoteProblemException problem) return problem; - } - return null; - } - - /** Problem status, which can differ from the HTTP envelope status. */ - public int getStatus() { return status; } - /** Remote error title, or an empty string. */ - public String getTitle() { return title; } - /** Remote business failure detail. */ - public String getDetail() { return detail; } - /** Remote problem type, preserved even when empty. */ - public String getType() { return type; } - /** Remote timestamp when supplied. */ - public String getTimestamp() { return timestamp; } -} diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java index a6465690..c57de42e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/FailureMapping.java @@ -19,12 +19,12 @@ package dev.openan.workflow.engine.core; -import dev.openan.workflow.engine.client.RemoteProblemException; +import dev.openan.workflow.engine.client.RemoteA2AErrorException; import dev.openan.workflow.engine.model.BusinessFailure; import dev.openan.workflow.engine.model.TaskResult; import java.util.*; -/** Maps generic failures, remote problem facts, or explicitly safe host-selected business facts. */ +/** Maps generic failures, standard A2A errors, or explicitly safe host-selected business facts. */ final class FailureMapping { static TaskResult from(Throwable error) { Throwable root = error; @@ -32,7 +32,7 @@ static TaskResult from(Throwable error) { while (root.getCause() != null && seen.add(root) && !(root instanceof BusinessFailure) - && !(root instanceof RemoteProblemException)) { + && !(root instanceof RemoteA2AErrorException)) { root = root.getCause(); } if (root instanceof BusinessFailure business) { @@ -43,18 +43,23 @@ static TaskResult from(Throwable error) { .errorDetails(business.details()) .build(); } - if (root instanceof RemoteProblemException problem) { + RemoteA2AErrorException remoteError = RemoteA2AErrorException.findIn(root); + if (remoteError != null) { + Map details = new LinkedHashMap<>(); + details.put("httpStatus", remoteError.getHttpStatus()); + details.put("code", remoteError.getCode()); + if (!remoteError.getStatus().isBlank()) details.put("status", remoteError.getStatus()); + if (!remoteError.getReason().isBlank()) details.put("reason", remoteError.getReason()); + if (!remoteError.getDomain().isBlank()) details.put("domain", remoteError.getDomain()); + if (!remoteError.getDetails().isEmpty()) details.put("details", remoteError.getDetails()); + if (!remoteError.getRetryAfter().isBlank()) { + details.put("retryAfter", remoteError.getRetryAfter()); + } return TaskResult.builder() .success(false) - .errorCode("remote.problem." + problem.getStatus()) - .error(problem.getMessage()) - .errorDetails( - Map.of( - "status", problem.getStatus(), - "title", problem.getTitle(), - "detail", problem.getDetail(), - "type", problem.getType(), - "timestamp", problem.getTimestamp())) + .errorCode(remoteError.workflowErrorCode()) + .error(remoteError.getMessage()) + .errorDetails(details) .build(); } String code = diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java similarity index 67% rename from workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java rename to workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java index c17a1ed7..c4ae3639 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProblemDetectingHttpClientTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java @@ -5,16 +5,16 @@ * 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 + * 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. + * 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; @@ -31,15 +31,15 @@ import org.a2aproject.sdk.client.http.ServerSentEvent; import org.junit.jupiter.api.Test; -class ProblemDetectingHttpClientTest { +class A2AErrorDetectingHttpClientTest { @Test - void synchronousProblemCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws Exception { + void standardErrorCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws Exception { var upstream = new CompletableFuture(); var failures = new AtomicInteger(); var events = new AtomicInteger(); var completions = new AtomicInteger(); var client = client((messages, errors, complete) -> { - messages.accept(new ServerSentEvent("{\"status\":429,\"detail\":\"capacity reached\"}")); + messages.accept(new ServerSentEvent(RemoteA2AErrorResponseTest.error(429))); messages.accept(new ServerSentEvent("{\"message\":{}}")); complete.run(); errors.accept(new IllegalStateException("late close")); @@ -50,10 +50,11 @@ void synchronousProblemCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws failures.incrementAndGet(); throw new IllegalStateException("observer failed"); }, completions::incrementAndGet); - var problem = RemoteProblemException.findIn(assertThrows(CompletionException.class, result::join)); - assertNotNull(problem); - assertEquals(429, problem.getStatus()); - assertEquals(1, problem.getSuppressed().length); + var remoteError = RemoteA2AErrorException.findIn( + assertThrows(CompletionException.class, result::join)); + assertNotNull(remoteError); + assertEquals(429, remoteError.getHttpStatus()); + assertEquals(1, remoteError.getSuppressed().length); assertTrue(upstream.isCancelled()); assertEquals(1, failures.get()); assertEquals(0, events.get()); @@ -80,8 +81,9 @@ void cancellationPropagatesAndLateEventsCannotCompleteTheCall() throws Exception } @Test - void successfulDataIsNotRewrittenOrMistakenForANestedProblem() throws Exception { - String data = "{\"task\":{\"status\":{\"state\":\"TASK_STATE_COMPLETED\"}},\"metadata\":{\"status\":400}}"; + void successfulDataIsNotRewrittenOrMistakenForANestedError() throws Exception { + String data = "{\"task\":{\"status\":{\"state\":\"TASK_STATE_COMPLETED\"}}," + + "\"metadata\":{\"error\":{\"code\":400,\"message\":\"business value\"}}}"; var seen = new AtomicReference(); var completions = new AtomicInteger(); var result = client((messages, errors, complete) -> { @@ -95,8 +97,25 @@ void successfulDataIsNotRewrittenOrMistakenForANestedProblem() throws Exception assertEquals(1, completions.get()); } + @Test + void streamThatCompletesWithoutAnA2AEventFailsImmediately() throws Exception { + var failures = new AtomicReference(); + var completions = new AtomicInteger(); + var result = client((messages, errors, complete) -> { + complete.run(); + return CompletableFuture.completedFuture(null); + }).createPost().postAsyncSSE(event -> fail("unexpected event"), failures::set, + completions::incrementAndGet); + + var failure = assertThrows(CompletionException.class, result::join); + assertInstanceOf(java.io.IOException.class, failure.getCause()); + assertEquals("A2A streaming response closed without an event", failure.getCause().getMessage()); + assertSame(failure.getCause(), failures.get()); + assertEquals(0, completions.get()); + } + private static A2AHttpClient client(Start start) { - return new ProblemDetectingHttpClient(new A2AHttpClient() { + return new A2AErrorDetectingHttpClient(new A2AHttpClient() { @Override public GetBuilder createGet() { throw new UnsupportedOperationException(); } @Override public DeleteBuilder createDelete() { throw new UnsupportedOperationException(); } @Override public PostBuilder createPost() { @@ -117,6 +136,7 @@ private static A2AHttpClient client(Start start) { @FunctionalInterface private interface Start { - CompletableFuture run(Consumer messages, Consumer errors, Runnable complete); + CompletableFuture run(Consumer messages, Consumer errors, + Runnable complete); } } 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 b8e94cca..5991209d 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 @@ -22,10 +22,12 @@ import static org.junit.jupiter.api.Assertions.*; import com.sun.net.httpserver.HttpServer; +import java.io.IOException; import java.net.*; import java.net.http.*; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.*; import java.util.concurrent.*; import org.junit.jupiter.api.AfterEach; @@ -177,4 +179,116 @@ void loggingFailureCannotChangeBodyDeliveryAndBodySwitchIsExplicit() throws Exce "", Map.of())); } + + @Test + void subscriberCancellationIsNotLoggedAsATransportFailure() { + List records = new CopyOnWriteArrayList<>(); + HttpRequest request = + HttpRequest.newBuilder(URI.create("https://agent.example/message:stream")).GET().build(); + HttpResponse.BodyHandler cancellingHandler = + ignored -> + new HttpResponse.BodySubscriber<>() { + private final CompletableFuture body = new CompletableFuture<>(); + + @Override + public CompletionStage getBody() { + return body; + } + + @Override + public void onSubscribe(Flow.Subscription subscription) { + subscription.cancel(); + } + + @Override + public void onNext(List item) {} + + @Override + public void onError(Throwable error) { + body.completeExceptionally(error); + } + + @Override + public void onComplete() { + body.complete(null); + } + }; + + CompletableFuture> result = + new ObservedHttpClient(new CancellationReportingHttpClient(), records::add) + .sendAsync(request, cancellingHandler); + + assertThrows(CompletionException.class, result::join); + assertTrue(records.stream().noneMatch(entry -> entry.direction().equals("FAILURE"))); + } + + private static final class CancellationReportingHttpClient extends HttpClient { + @Override + public HttpResponse send(HttpRequest request, HttpResponse.BodyHandler handler) { + throw new UnsupportedOperationException(); + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, HttpResponse.BodyHandler handler) { + CompletableFuture> result = new CompletableFuture<>(); + HttpResponse.ResponseInfo info = + new HttpResponse.ResponseInfo() { + @Override + public int statusCode() { + return 200; + } + + @Override + public HttpHeaders headers() { + return HttpHeaders.of( + Map.of("Content-Type", List.of("text/event-stream")), (name, value) -> true); + } + + @Override + public Version version() { + return Version.HTTP_1_1; + } + }; + handler + .apply(info) + .onSubscribe( + new Flow.Subscription() { + @Override + public void request(long count) {} + + @Override + public void cancel() { + result.completeExceptionally(new IOException("subscriber cancelled")); + } + }); + return result; + } + + @Override + public CompletableFuture> sendAsync( + HttpRequest request, + HttpResponse.BodyHandler handler, + HttpResponse.PushPromiseHandler pushPromiseHandler) { + return sendAsync(request, handler); + } + + @Override public Optional cookieHandler() { return Optional.empty(); } + @Override public Optional connectTimeout() { return Optional.empty(); } + @Override public Redirect followRedirects() { return Redirect.NEVER; } + @Override public Optional proxy() { return Optional.empty(); } + @Override public javax.net.ssl.SSLContext sslContext() { return defaultSslContext(); } + @Override public javax.net.ssl.SSLParameters sslParameters() { return new javax.net.ssl.SSLParameters(); } + @Override public Optional authenticator() { return Optional.empty(); } + @Override public Version version() { return Version.HTTP_1_1; } + @Override public Optional executor() { return Optional.empty(); } + + private static javax.net.ssl.SSLContext defaultSslContext() { + try { + return javax.net.ssl.SSLContext.getDefault(); + } catch (java.security.NoSuchAlgorithmException error) { + throw new IllegalStateException(error); + } + } + } } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java new file mode 100644 index 00000000..c157baf4 --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteA2AErrorResponseTest.java @@ -0,0 +1,194 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.*; + +import com.sun.net.httpserver.HttpServer; +import java.net.InetSocketAddress; +import java.net.http.HttpClient; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.a2aproject.sdk.client.http.JdkA2AHttpClient; +import org.a2aproject.sdk.spec.*; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +class RemoteA2AErrorResponseTest { + static String error(int code) { + String status = code == 429 ? "RESOURCE_EXHAUSTED" : "INVALID_ARGUMENT"; + String reason = code == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "INVALID_PARAMS"; + String domain = code == 429 ? "fixture.invalid" : "a2a-protocol.org"; + return "{\"error\":{\"code\":" + code + ",\"status\":\"" + status + + "\",\"message\":\"该端口下未发现业务信息\",\"details\":[" + + "{\"@type\":\"type.googleapis.com/google.rpc.BadRequest\"," + + "\"fieldViolations\":[{\"field\":\"port\",\"description\":\"required\"}]}," + + "{\"@type\":\"type.googleapis.com/google.rpc.ErrorInfo\",\"reason\":\"" + + reason + "\",\"domain\":\"" + domain + "\"," + + "\"metadata\":{\"taskId\":\"task-123\"}}]}}"; + } + + @Test + void preservesStandardFieldsAndFindsErrorInfoAmongTypedDetails() { + var error = RemoteA2AErrorException.fromResponse( + 400, error(400), Map.of("Retry-After", List.of("5"))); + assertNotNull(error); + assertEquals(400, error.getHttpStatus()); + assertEquals(400, error.getCode()); + assertEquals("INVALID_ARGUMENT", error.getStatus()); + assertEquals("INVALID_PARAMS", error.getReason()); + assertEquals("a2a-protocol.org", error.getDomain()); + assertEquals("该端口下未发现业务信息", error.getMessage()); + assertEquals(2, error.getDetails().size()); + assertEquals("5", error.getRetryAfter()); + assertEquals("a2a.invalid_params", error.workflowErrorCode()); + } + + @Test + void ignoresOldProblemShapeAndNestedBusinessErrors() { + assertNull(RemoteA2AErrorException.fromPayload( + "{\"status\":400,\"title\":\"Invalid Params\",\"detail\":\"missing\"}")); + assertNull(RemoteA2AErrorException.fromPayload( + "{\"artifactUpdate\":{\"artifact\":{\"parts\":[{\"data\":" + error(400) + "}]}}}")); + assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{\"code\":200,\"message\":\"ok\"}}")); + assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{\"code\":400}}")); + assertNull(RemoteA2AErrorException.fromPayload("{\"error\":{")); + } + + @Test + void redactsTypedDetailsAndResponseHeaders() { + String payload = error(400).replace("\"taskId\":\"task-123\"", + "\"accessSession\":\"secret-value\""); + var error = RemoteA2AErrorException.fromResponse( + 400, payload, Map.of("Authorization", List.of("Bearer secret"))); + assertEquals("***", ((Map) error.getDetails().get(1).get("metadata")).get("accessSession")); + assertEquals(List.of("***"), error.getResponseHeaders().get("Authorization")); + } + + @Test + void normalizesPeerSuppliedReasonBeforeUsingItAsAWorkflowCode() { + String payload = + error(400).replace("INVALID_PARAMS", "Invalid Params / Port#1"); + + var error = RemoteA2AErrorException.fromPayload(payload); + + assertNotNull(error); + assertEquals("a2a.invalid_params_port_1", error.workflowErrorCode()); + } + + @ParameterizedTest + @ValueSource(ints = {400, 429, 503}) + void directRuntimeFailsPromptlyWhenSdkExposesAnErrorBodyAsSseData(int status) throws Exception { + var release = new CountDownLatch(1); + var disconnected = new CountDownLatch(1); + var requests = new AtomicInteger(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + requests.incrementAndGet(); + exchange.getRequestBody().readAllBytes(); + exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, 0); + try { + byte[] body = ("data: " + error(status) + "\n\n").getBytes(StandardCharsets.UTF_8); + exchange.getResponseBody().write(body, 0, 12); + exchange.getResponseBody().flush(); + exchange.getResponseBody().write(body, 12, body.length - 12); + exchange.getResponseBody().flush(); + for (int i = 0; i < 80 && !release.await(100, TimeUnit.MILLISECONDS); i++) { + exchange.getResponseBody().write(": keep-alive\n\n".getBytes(StandardCharsets.UTF_8)); + exchange.getResponseBody().flush(); + } + } catch (java.io.IOException clientClosed) { + disconnected.countDown(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + exchange.close(); + } + }); + server.start(); + var runtime = new DefaultA2AJavaClientRuntime(true, null, 20, "HTTP+JSON"); + try { + AgentCard card = AgentCard.builder().name("remote-test").description("test").version("1") + .capabilities(AgentCapabilities.builder().streaming(true).build()) + .defaultInputModes(List.of("text/plain")).defaultOutputModes(List.of("text/plain")) + .skills(List.of()).supportedInterfaces(List.of(new AgentInterface("HTTP+JSON", + "http://127.0.0.1:" + server.getAddress().getPort() + "/a2a/json"))).build(); + MessageSendParams params = MessageSendParams.builder().message(Message.builder() + .role(Message.Role.ROLE_USER).messageId("test").parts(new TextPart("diagnose")).build()).build(); + var events = new AtomicInteger(); + var response = CompletableFuture.supplyAsync( + () -> runtime.sendMessage(card, params, null, event -> events.incrementAndGet(), null)); + var thrown = assertThrows(java.util.concurrent.ExecutionException.class, + () -> response.get(3, TimeUnit.SECONDS)); + var remoteError = RemoteA2AErrorException.findIn(thrown); + assertNotNull(remoteError); + assertEquals(status, remoteError.getHttpStatus()); + assertEquals(0, events.get()); + assertEquals(1, requests.get()); + assertTrue(disconnected.await(3, TimeUnit.SECONDS), "failed SSE must close its HTTP connection"); + } finally { + release.countDown(); + runtime.close(); + server.stop(0); + } + } + + @Test + void projectsTypedSdkErrorsAndHandlesCauseCycles() { + var projected = RemoteA2AErrorException.findIn( + new IllegalStateException("wrapper", new InvalidParamsError("missing port"))); + assertNotNull(projected); + assertEquals("INVALID_PARAMS", projected.getReason()); + assertEquals("a2a.invalid_params", projected.workflowErrorCode()); + assertNull(RemoteA2AErrorException.findIn(null)); + var one = new IllegalStateException("one"); + var two = new IllegalStateException("two", one); + one.initCause(two); + assertNull(RemoteA2AErrorException.findIn(one)); + } + + @Test + void nonStreamingResponsesPreserveTheStandardA2AError() throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + byte[] body = error(400).getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/a2a+json"); + exchange.sendResponseHeaders(400, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + server.start(); + try { + var client = new A2AErrorDetectingHttpClient(new JdkA2AHttpClient(HttpClient.newHttpClient())); + var error = assertThrows(RemoteA2AErrorException.class, () -> client.createPost() + .url("http://127.0.0.1:" + server.getAddress().getPort()).body("{}").post()); + assertEquals(400, error.getHttpStatus()); + assertEquals("INVALID_PARAMS", error.getReason()); + } finally { + server.stop(0); + } + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java deleted file mode 100644 index 4d1caba8..00000000 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RemoteProblemResponseTest.java +++ /dev/null @@ -1,152 +0,0 @@ -/* - * 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 static org.junit.jupiter.api.Assertions.*; - -import com.sun.net.httpserver.HttpServer; -import java.net.InetSocketAddress; -import java.net.http.HttpClient; -import java.nio.charset.StandardCharsets; -import java.util.List; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import org.a2aproject.sdk.client.http.JdkA2AHttpClient; -import org.a2aproject.sdk.spec.*; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; - -class RemoteProblemResponseTest { - static String problem(int status) { - return "{\"status\":" + status + ",\"detail\":\"该端口下未发现业务信息\"," - + "\"title\":\"Invalid Params\",\"type\":\"\",\"timestamp\":\"2026-08-31T09:07:36Z\"}"; - } - - @Test - void preservesProblemFieldsWithoutInspectingNestedBusinessData() { - var error = RemoteProblemException.fromPayload(problem(400)); - assertNotNull(error); - assertEquals(400, error.getStatus()); - assertEquals("Invalid Params", error.getTitle()); - assertEquals("该端口下未发现业务信息", error.getDetail()); - assertEquals("", error.getType()); - assertEquals("2026-08-31T09:07:36Z", error.getTimestamp()); - assertNull(RemoteProblemException.fromPayload("{\"artifactUpdate\":{\"data\":" + problem(400) + "}}")); - assertNull(RemoteProblemException.fromPayload("{\"status\":{\"state\":\"TASK_STATE_WORKING\"}}")); - assertNull(RemoteProblemException.fromPayload("{\"status\":400}")); - assertNull(RemoteProblemException.fromPayload(problem(200))); - assertNull(RemoteProblemException.fromPayload("{\"status\":429,")); - } - - @ParameterizedTest - @ValueSource(ints = {400, 429, 503}) - void directRuntimeFailsPromptlyWithoutWaitingForSseClosureOrRetrying(int status) throws Exception { - var release = new CountDownLatch(1); - var disconnected = new CountDownLatch(1); - var requests = new AtomicInteger(); - HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); - server.createContext("/", exchange -> { - requests.incrementAndGet(); - exchange.getRequestBody().readAllBytes(); - exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); - exchange.sendResponseHeaders(200, 0); - try { - String payload = status == 503 - ? problem(status).replace("该端口下未发现业务信息", "connection closed locally; request rejected") - : problem(status); - byte[] body = ("data: " + payload + "\n\n").getBytes(StandardCharsets.UTF_8); - exchange.getResponseBody().write(body, 0, 12); - exchange.getResponseBody().flush(); - exchange.getResponseBody().write(body, 12, body.length - 12); - exchange.getResponseBody().flush(); - for (int i = 0; i < 80 && !release.await(100, TimeUnit.MILLISECONDS); i++) { - exchange.getResponseBody().write(": keep-alive\n\n".getBytes(StandardCharsets.UTF_8)); - exchange.getResponseBody().flush(); - } - } catch (java.io.IOException clientClosed) { - disconnected.countDown(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - exchange.close(); - } - }); - server.start(); - var runtime = new DefaultA2AJavaClientRuntime(true, null, 20, "HTTP+JSON"); - try { - AgentCard card = AgentCard.builder().name("remote-test").description("test").version("1") - .capabilities(AgentCapabilities.builder().streaming(true).build()) - .defaultInputModes(List.of("text/plain")).defaultOutputModes(List.of("text/plain")) - .skills(List.of()).supportedInterfaces(List.of(new AgentInterface("HTTP+JSON", - "http://127.0.0.1:" + server.getAddress().getPort() + "/a2a/json"))).build(); - MessageSendParams params = MessageSendParams.builder().message(Message.builder() - .role(Message.Role.ROLE_USER).messageId("test").parts(new TextPart("diagnose")).build()).build(); - var events = new AtomicInteger(); - var response = CompletableFuture.supplyAsync( - () -> runtime.sendMessage(card, params, null, event -> events.incrementAndGet(), null)); - var thrown = assertThrows(java.util.concurrent.ExecutionException.class, - () -> response.get(3, TimeUnit.SECONDS)); - Throwable cause = thrown; - while (cause.getCause() != null && !(cause instanceof RemoteProblemException)) cause = cause.getCause(); - var problem = assertInstanceOf(RemoteProblemException.class, cause); - assertEquals(status, problem.getStatus()); - assertEquals(0, events.get()); - assertEquals(1, requests.get()); - assertTrue(disconnected.await(3, TimeUnit.SECONDS), "failed SSE must close its HTTP connection"); - } finally { - release.countDown(); - runtime.close(); - server.stop(0); - } - } - - @Test - void causeLookupHandlesNullCyclesAndWrappedProblems() { - var problem = RemoteProblemException.fromPayload(problem(429)); - assertSame(problem, RemoteProblemException.findIn(new IllegalStateException("wrapper", problem))); - assertNull(RemoteProblemException.findIn(null)); - var one = new IllegalStateException("one"); - var two = new IllegalStateException("two", one); - one.initCause(two); - assertNull(RemoteProblemException.findIn(one)); - } - - @Test - void nonStreamingResponsesPreserveTheRemoteBusinessFailure() throws Exception { - HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); - server.createContext("/", exchange -> { - byte[] body = problem(400).getBytes(StandardCharsets.UTF_8); - exchange.sendResponseHeaders(400, body.length); - exchange.getResponseBody().write(body); - exchange.close(); - }); - server.start(); - try { - var client = new ProblemDetectingHttpClient(new JdkA2AHttpClient(HttpClient.newHttpClient())); - var error = assertThrows(RemoteProblemException.class, () -> client.createPost() - .url("http://127.0.0.1:" + server.getAddress().getPort()).body("{}").post()); - assertEquals(400, error.getStatus()); - } finally { - server.stop(0); - } - } -} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java similarity index 61% rename from workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java rename to workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java index ac07d5a5..78ac0e60 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteProblemFailureTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java @@ -5,23 +5,23 @@ * 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 + * 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. + * 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.core; import static org.junit.jupiter.api.Assertions.*; import dev.openan.workflow.engine.StubWorkflowEngineClient; -import dev.openan.workflow.engine.client.RemoteProblemException; +import dev.openan.workflow.engine.client.RemoteA2AErrorException; import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.model.*; import java.util.List; @@ -33,31 +33,41 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; -class RemoteProblemFailureTest { +class RemoteA2AErrorFailureTest { + private static String error(int status) { + String canonical = status == 429 ? "RESOURCE_EXHAUSTED" : "INVALID_ARGUMENT"; + String reason = status == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "INVALID_PARAMS"; + return "{\"error\":{\"code\":" + status + ",\"status\":\"" + canonical + + "\",\"message\":\"Remote agent rejected the request\",\"details\":[{" + + "\"@type\":\"type.googleapis.com/google.rpc.ErrorInfo\",\"reason\":\"" + + reason + "\",\"domain\":\"example.invalid\"}]}}"; + } + @ParameterizedTest @ValueSource(ints = {400, 429}) - void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status) throws Exception { + void waitsForTheOtherAgentAndSkipsMergeWhileReportingCorrelatedFailure(int status) + throws Exception { var started = new java.util.concurrent.CountDownLatch(2); var reported = new java.util.concurrent.CountDownLatch(1); - var city1 = new CompletableFuture(); - var city2 = new CompletableFuture(); + var agent1 = new CompletableFuture(); + 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) { started.countDown(); - return request.getAgentName().equals("city1") ? city1 : city2; + return request.getAgentName().equals("agent1") ? agent1 : agent2; } }; var merges = new AtomicInteger(); var callbacks = ControlPoint.builder() - .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("diagnose"))) + .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("run task"))) .onSelfTask(request -> { merges.incrementAndGet(); return CompletableFuture.completedFuture(TaskResult.success(List.of("must not merge"))); }).build(); var workflow = Workflow.builder().name("parallel-failure").steps(List.of( - city("city1"), city("city2"), + remote("agent1"), remote("agent2"), WorkflowStep.builder().name("merge").stepType(StepType.SELF_LOOP) .subtasks(List.of(Task.builder().agent("host").description("aggregate").build())) .next(List.of()).build())).build(); @@ -72,17 +82,16 @@ void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status } } }; - var execution = new WorkflowExecutor(workflow, callbacks, client, events, "complaint", "zh").run(); + var execution = new WorkflowExecutor(workflow, callbacks, client, events, "request", "en").run(); try { assertTrue(started.await(5, java.util.concurrent.TimeUnit.SECONDS)); - city1.completeExceptionally(RemoteProblemException.fromPayload( - "{\"status\":" + status + ",\"detail\":\"OMC拒绝请求\"}")); + agent1.completeExceptionally(RemoteA2AErrorException.fromPayload(error(status))); assertTrue(reported.await(5, java.util.concurrent.TimeUnit.SECONDS)); - assertFalse(execution.isDone(), "The other in-flight diagnosis must be collected"); - assertFalse(city2.isCancelled()); - city2.complete(SendMessageResult.builder().taskState("TASK_STATE_COMPLETED") + assertFalse(execution.isDone(), "The other in-flight task must be collected"); + assertFalse(agent2.isCancelled()); + agent2.complete(SendMessageResult.builder().taskState("TASK_STATE_COMPLETED") .receivedMessages(List.of(new ReceivedMessage( - MessageContent.text("city2 diagnosis"), Map.of(), List.of()))).build()); + MessageContent.text("agent2 result"), Map.of(), List.of()))).build()); var result = execution.get(5, java.util.concurrent.TimeUnit.SECONDS); assertFalse(result.isSuccess()); assertEquals(2, result.getHistory().size()); @@ -95,67 +104,68 @@ void waitsForTheOtherCityAndSkipsMergeWhileReportingCorrelatedFailure(int status } assertEquals(observed.get(0).get("executionId"), observed.get(1).get("executionId")); assertNotEquals(observed.get(0).get("taskId"), observed.get(1).get("taskId")); - var success = result.getHistory().stream().filter(h -> h.get("agent").equals("city2")).findFirst().orElseThrow(); + var success = result.getHistory().stream() + .filter(h -> h.get("agent").equals("agent2")).findFirst().orElseThrow(); assertEquals("success", success.get("status")); - assertEquals(List.of("city2 diagnosis"), success.get("outputs")); + assertEquals(List.of("agent2 result"), success.get("outputs")); } finally { - city1.cancel(true); - city2.cancel(true); + agent1.cancel(true); + agent2.cancel(true); execution.cancel(true); } } - private static WorkflowStep city(String name) { + private static WorkflowStep remote(String name) { return WorkflowStep.builder().name(name).stepType(StepType.ALL_SUCCESS) - .subtasks(List.of(Task.builder().agent(name).description("diagnose").build())) + .subtasks(List.of(Task.builder().agent(name).description("execute").build())) .next(List.of(new JumpCondition("merge", ""))).build(); } + @ParameterizedTest @ValueSource(ints = {400, 429}) - void remoteFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs(int status) { - var problem = RemoteProblemException.fromPayload( - "{\"status\":" + status + ",\"detail\":\"OMC拒绝请求\",\"type\":\"\"}"); + void protocolFailureReachesWorkflowHistoryWithoutNegotiationOrSuccessfulOutputs(int status) { + var remoteError = RemoteA2AErrorException.fromPayload(error(status)); var calls = new AtomicInteger(); var negotiations = new AtomicInteger(); - var client = new StubWorkflowEngineClient("omc") { + var client = new StubWorkflowEngineClient("remote-agent") { @Override public CompletableFuture dispatch( TaskRequest request, MessageContent content, ControlPoint callbacks) { calls.incrementAndGet(); - return CompletableFuture.failedFuture(new CompletionException(problem)); + return CompletableFuture.failedFuture(new CompletionException(remoteError)); } }; - var step = WorkflowStep.builder().name("diagnose").stepType(StepType.ALL_SUCCESS) - .subtasks(List.of(Task.builder().agent("omc").description("diagnose").build())) + var step = WorkflowStep.builder().name("remote").stepType(StepType.ALL_SUCCESS) + .subtasks(List.of(Task.builder().agent("remote-agent").description("execute").build())) .next(List.of()).build(); var workflow = Workflow.builder().name("remote-failure").steps(List.of(step)).build(); var callbacks = ControlPoint.builder() - .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("diagnose"))) + .onTask(request -> CompletableFuture.completedFuture(MessageContent.text("execute"))) .onNegotiation(request -> { negotiations.incrementAndGet(); return CompletableFuture.failedFuture(new AssertionError("must not negotiate")); }).build(); - var result = new WorkflowExecutor(workflow, callbacks, client, null, "complaint", "zh").run().join(); + var result = new WorkflowExecutor(workflow, callbacks, client, null, "request", "en").run().join(); assertFalse(result.isSuccess()); assertEquals(1, calls.get()); assertEquals(0, negotiations.get()); var history = result.getHistory().get(0); assertEquals("failed", history.get("status")); - assertEquals("remote.problem." + status, history.get("errorCode")); + assertEquals(status == 429 ? "a2a.active_task_limit_exceeded" : "a2a.invalid_params", + history.get("errorCode")); assertEquals(List.of(), history.get("outputs")); - assertTrue(history.get("error").toString().contains("OMC拒绝请求")); - assertEquals(status, ((Map) history.get("errorDetails")).get("status")); - assertEquals("", ((Map) history.get("errorDetails")).get("type")); + assertTrue(history.get("error").toString().contains("rejected")); + assertEquals(status, ((Map) history.get("errorDetails")).get("httpStatus")); + assertNotNull(((Map) history.get("errorDetails")).get("details")); } @Test void knownSecretsAreRedactedAndUnrecognizedProviderErrorsRemainPrivate() { - var problem = RemoteProblemException.fromPayload( - "{\"status\":400,\"detail\":\"bad request password=private-value accessSession=private-session\"}"); - var result = FailureMapping.from(problem); + var remoteError = RemoteA2AErrorException.fromPayload(error(400).replace( + "Remote agent rejected the request", "bad request password=private-value")); + var result = FailureMapping.from(remoteError); assertFalse(result.toString().contains("private-value")); - assertFalse(result.toString().contains("private-session")); - assertFalse(problem.getMessage().contains("private-value")); + assertFalse(remoteError.getMessage().contains("private-value")); assertEquals("IllegalStateException", FailureMapping.from(new IllegalStateException("raw private provider body")).getError()); } From 5ce9353ff3e401dd8dfd29c95253553131793538 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Fri, 4 Sep 2026 20:35:46 +0800 Subject: [PATCH 2/4] fix(starter): stream SseEmitter directly and map pre-stream A2A errors A wildcard ResponseEntity return bypassed Spring's emitter handler, so every message:stream and task subscription response degraded to an HTTP 500 with an empty body; the client-side SDK then treated the zero-event completion as success and callers waited out the send timeout. Declare SseEmitter as the return type and throw A2AError for pre-stream failures, mapped by an @ExceptionHandler to a non-2xx application/a2a+json standard envelope. Enforce the SDK pre-contracts in order (streaming capability, protocol version, required extensions, task access) before dispatch. Release the Flow subscription and notify the SDK event consumer on completion, error, timeout and write failure. Declare text/event-stream produces on the streaming mappings and return application/a2a+json on protocol responses. The negative tests now go through MockMvc dispatch, which is what would have caught the 500 regression. Signed-off-by: Zhoujie628 --- .../engine/spring/A2AAutoConfiguration.java | 8 +- .../workflow/engine/spring/A2AController.java | 269 ++++++++++-------- .../spring/A2AAutoConfigurationTest.java | 156 ++++++++++ 3 files changed, 307 insertions(+), 126 deletions(-) diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java index 2688cd2c..03514a6a 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AAutoConfiguration.java @@ -176,8 +176,7 @@ public MainEventBusProcessor eventBusProcessor( PushNotificationConfigStore pushStore) { PushNotificationSender sender = new BasePushNotificationSender(pushStore); MainEventBusProcessor proc = new MainEventBusProcessor(bus, store, sender, qm); - proc.ensureStarted(); - log.info("[A2A] MainEventBusProcessor started"); + log.info("[A2A] MainEventBusProcessor configured; Spring manages its lifecycle"); return proc; } @@ -211,7 +210,8 @@ public RestHandler restHandler(AgentCard card, RequestHandler handler, ExecutorS @Bean @ConditionalOnMissingBean - public A2AController a2aController(RestHandler restHandler, RequestHandler requestHandler) { - return new A2AController(restHandler, requestHandler); + public A2AController a2aController( + RestHandler restHandler, RequestHandler requestHandler, AgentCard agentCard) { + return new A2AController(restHandler, requestHandler, agentCard); } } diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java index 132242e2..d5ccc15d 100644 --- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java +++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AController.java @@ -31,18 +31,29 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import org.a2aproject.sdk.grpc.SendMessageRequest; import org.a2aproject.sdk.grpc.StreamResponse; import org.a2aproject.sdk.grpc.utils.ProtoUtils; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.auth.TaskOperation; +import org.a2aproject.sdk.server.extensions.A2AExtensions; import org.a2aproject.sdk.server.requesthandlers.RequestHandler; +import org.a2aproject.sdk.server.version.A2AVersionValidator; +import org.a2aproject.sdk.spec.A2AError; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.InternalError; +import org.a2aproject.sdk.spec.InvalidRequestError; import org.a2aproject.sdk.spec.MessageSendParams; import org.a2aproject.sdk.spec.StreamingEventKind; +import org.a2aproject.sdk.spec.TaskIdParams; +import org.a2aproject.sdk.spec.UnsupportedOperationError; import org.a2aproject.sdk.transport.rest.handler.RestHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; @@ -69,94 +80,149 @@ *

The path prefix is configurable via {@code a2at.server.path-prefix}. The controller delegates * to {@link RestHandler} (non-streaming) and {@link RequestHandler} (streaming). The {@link * ServerCallContext} is built from {@link HttpServletRequest} headers, preserving the - * A2A-Extensions header. + * A2A-Extensions header. Streaming endpoints return {@link SseEmitter}; a request rejected before + * the stream is created throws {@link A2AError}, which {@link #a2aError(A2AError)} maps to a + * non-2xx response with the standard A2A error envelope. */ @RestController public class A2AController { private static final Logger log = LoggerFactory.getLogger(A2AController.class); + private static final MediaType A2A_JSON = MediaType.parseMediaType("application/a2a+json"); private final RestHandler restHandler; private final RequestHandler requestHandler; + private final AgentCard agentCard; private final AtomicInteger activeStreams = new AtomicInteger(); private final Object streamCompletionMonitor = new Object(); - public A2AController(RestHandler restHandler, RequestHandler requestHandler) { + public A2AController( + RestHandler restHandler, RequestHandler requestHandler, AgentCard agentCard) { this.restHandler = restHandler; this.requestHandler = requestHandler; + this.agentCard = agentCard; } @PostMapping("${a2at.server.path-prefix}/message:send") public ResponseEntity sendMessage(HttpServletRequest req, @RequestBody String body) { var ctx = buildContext(req); var resp = restHandler.sendMessage(ctx, "", body); - return ResponseEntity.status(resp.getStatusCode()) - .contentType( - MediaType.parseMediaType( - resp.getContentType() != null - ? resp.getContentType() - : MediaType.APPLICATION_JSON_VALUE)) - .body(resp.getBody()); + return toResponse(resp); } @PostMapping( value = "${a2at.server.path-prefix}/message:stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter streamMessage(HttpServletRequest req, @RequestBody String body) { - SseEmitter emitter = track(new SseEmitter(0L)); try { + var ctx = buildContext(req); + requireStreamingSupport(); + A2AVersionValidator.validateProtocolVersion(agentCard, ctx); + A2AExtensions.validateRequiredExtensions(agentCard, ctx); + SendMessageRequest.Builder builder = SendMessageRequest.newBuilder(); JsonFormat.parser().merge(body, builder); MessageSendParams params = ProtoUtils.FromProto.messageSendParams(builder.build()); - - var ctx = buildContext(req); + requestHandler.authorizeTaskAccess( + params.message().taskId(), ctx, TaskOperation.MESSAGE_SEND_STREAM); Flow.Publisher publisher = requestHandler.onMessageSendStream(params, ctx); + return subscribeToEmitter(publisher, ctx); + } catch (A2AError e) { + log.warn("[SSE] A2A request rejected before stream creation: {}", e.getMessage()); + throw e; + } catch (com.google.protobuf.InvalidProtocolBufferException e) { + log.warn("[SSE] Invalid A2A request body: {}", e.getMessage()); + throw new InvalidRequestError("Invalid A2A request body"); + } catch (Exception e) { + log.error("[SSE] Setup failed before stream creation: {}", e.getClass().getSimpleName(), e); + throw new InternalError("Streaming request setup failed"); + } + } - final AtomicLong seq = new AtomicLong(0); - publisher.subscribe( - new Flow.Subscriber<>() { - private Flow.Subscription sub; - - @Override - public void onSubscribe(Flow.Subscription s) { - sub = s; - s.request(1); - } + /** + * Maps a pre-stream {@link A2AError} to a non-2xx response with the standard A2A error envelope. + * + *

Must be an exception handler rather than a {@code ResponseEntity} return so that streaming + * endpoints can declare {@link SseEmitter} as their return type. A wildcard {@code + * ResponseEntity} return is not picked up by Spring's emitter handler and the emitter body is + * rejected by the message converters, turning every stream into a 500 response. + */ + @ExceptionHandler(A2AError.class) + ResponseEntity a2aError(A2AError error) { + var response = restHandler.createErrorResponse(error); + return toResponse(response); + } - @Override - public void onNext(StreamingEventKind item) { + private SseEmitter subscribeToEmitter( + Flow.Publisher publisher, ServerCallContext context) { + AtomicReference subscription = new AtomicReference<>(); + AtomicBoolean released = new AtomicBoolean(); + Runnable release = + () -> { + if (released.compareAndSet(false, true)) { + Flow.Subscription active = subscription.get(); + if (active != null) { try { - StreamResponse sr = ProtoUtils.ToProto.streamResponse(item); - String compact = JsonFormat.printer().omittingInsignificantWhitespace().print(sr); - emitter.send( - SseEmitter.event() - .id(Long.toString(seq.incrementAndGet())) - .data(compact, MediaType.APPLICATION_JSON)); - } catch (Exception e) { - log.error("[SSE] Write failed: {}", e.getMessage()); - sub.cancel(); - emitter.completeWithError(e); - return; + active.cancel(); + } catch (RuntimeException error) { + log.warn("[SSE] Failed to cancel publisher subscription: {}", error.getMessage()); } - sub.request(1); } - - @Override - public void onError(Throwable t) { - log.error("[SSE] Stream error: {}", t.getMessage()); - emitter.completeWithError(t); + try { + context.invokeEventConsumerCancelCallback(); + } catch (RuntimeException error) { + log.warn("[SSE] Failed to notify SDK event consumer cancellation: {}", error.getMessage()); } + } + }; + SseEmitter emitter = track(new SseEmitter(0L), release); + final AtomicLong seq = new AtomicLong(0); + publisher.subscribe( + new Flow.Subscriber<>() { + @Override + public void onSubscribe(Flow.Subscription s) { + if (!subscription.compareAndSet(null, s) || released.get()) { + s.cancel(); + } else { + s.request(1); + } + } - @Override - public void onComplete() { - emitter.complete(); + @Override + public void onNext(StreamingEventKind item) { + try { + StreamResponse sr = ProtoUtils.ToProto.streamResponse(item); + String compact = JsonFormat.printer().omittingInsignificantWhitespace().print(sr); + emitter.send( + SseEmitter.event() + .id(Long.toString(seq.incrementAndGet())) + .data(compact, MediaType.APPLICATION_JSON)); + } catch (Exception e) { + log.error("[SSE] Write failed: {}", e.getMessage()); + release.run(); + emitter.completeWithError(e); + return; } - }); - } catch (Exception e) { - log.error("[SSE] Setup failed: {}", e.getMessage(), e); - emitter.completeWithError(e); - } + Flow.Subscription active = subscription.get(); + if (active != null && !released.get()) { + active.request(1); + } + } + + @Override + public void onError(Throwable t) { + log.error("[SSE] Stream error: {}", t.getMessage()); + release.run(); + emitter.completeWithError(t); + } + + @Override + public void onComplete() { + release.run(); + emitter.complete(); + } + }); return emitter; } @@ -164,13 +230,7 @@ public void onComplete() { public ResponseEntity getTask(HttpServletRequest req, @PathVariable("id") String taskId) { var ctx = buildContext(req); var resp = restHandler.getTask(ctx, "", taskId, null); - return ResponseEntity.status(resp.getStatusCode()) - .contentType( - MediaType.parseMediaType( - resp.getContentType() != null - ? resp.getContentType() - : MediaType.APPLICATION_JSON_VALUE)) - .body(resp.getBody()); + return toResponse(resp); } @GetMapping("${a2at.server.path-prefix}/tasks") @@ -194,13 +254,7 @@ public ResponseEntity listTasks( historyLength, statusTimestampAfter, includeArtifacts); - return ResponseEntity.status(resp.getStatusCode()) - .contentType( - MediaType.parseMediaType( - resp.getContentType() != null - ? resp.getContentType() - : MediaType.APPLICATION_JSON_VALUE)) - .body(resp.getBody()); + return toResponse(resp); } @PostMapping("${a2at.server.path-prefix}/tasks/{id}:cancel") @@ -210,73 +264,37 @@ public ResponseEntity cancelTask( @RequestBody(required = false) String body) { var ctx = buildContext(req); var resp = restHandler.cancelTask(ctx, "", body != null ? body : "{}", taskId); - return ResponseEntity.status(resp.getStatusCode()) - .contentType( - MediaType.parseMediaType( - resp.getContentType() != null - ? resp.getContentType() - : MediaType.APPLICATION_JSON_VALUE)) - .body(resp.getBody()); + return toResponse(resp); } @PostMapping( value = "${a2at.server.path-prefix}/tasks/{id}:subscribe", produces = MediaType.TEXT_EVENT_STREAM_VALUE) - public SseEmitter subscribeToTask(HttpServletRequest req, @PathVariable("id") String taskId) { - SseEmitter emitter = track(new SseEmitter(0L)); + public SseEmitter subscribeToTask( + HttpServletRequest req, @PathVariable("id") String taskId) { try { var ctx = buildContext(req); - var resp = restHandler.subscribeToTask(ctx, "", taskId); - if (resp instanceof RestHandler.HTTPRestStreamingResponse streaming) { - final AtomicLong seq = new AtomicLong(0); - streaming - .getPublisher() - .subscribe( - new Flow.Subscriber() { - private Flow.Subscription sub; - - @Override - public void onSubscribe(Flow.Subscription s) { - sub = s; - s.request(1); - } - - @Override - public void onNext(String item) { - try { - emitter.send( - SseEmitter.event() - .id(Long.toString(seq.incrementAndGet())) - .data(item, MediaType.APPLICATION_JSON)); - } catch (Exception e) { - log.error("[SSE] Subscribe write failed: {}", e.getMessage()); - sub.cancel(); - emitter.completeWithError(e); - return; - } - sub.request(1); - } - - @Override - public void onError(Throwable t) { - log.error("[SSE] Subscribe stream error: {}", t.getMessage()); - emitter.completeWithError(t); - } - - @Override - public void onComplete() { - emitter.complete(); - } - }); - } else { - emitter.send(SseEmitter.event().id("1").data(resp.getBody(), MediaType.APPLICATION_JSON)); - emitter.complete(); - } + requireStreamingSupport(); + requestHandler.authorizeTaskAccess(taskId, ctx, TaskOperation.SUBSCRIBE_TO_TASK); + Flow.Publisher publisher = + requestHandler.onSubscribeToTask(TaskIdParams.builder().id(taskId).tenant("").build(), ctx); + return subscribeToEmitter(publisher, ctx); + } catch (A2AError e) { + log.warn("[SSE] Task subscription rejected before stream creation: {}", e.getMessage()); + throw e; } catch (Exception e) { - log.error("[SSE] Subscribe setup failed: {}", e.getMessage(), e); - emitter.completeWithError(e); + log.error( + "[SSE] Subscribe setup failed before stream creation: {}", + e.getClass().getSimpleName(), + e); + throw new InternalError("Task subscription setup failed"); + } + } + + private void requireStreamingSupport() throws UnsupportedOperationError { + if (!agentCard.capabilities().streaming()) { + throw new UnsupportedOperationError(null, "Streaming is not supported by the agent", null); } - return emitter; } /** @@ -313,12 +331,13 @@ public int activeStreamCount() { return activeStreams.get(); } - private T track(T emitter) { + private T track(T emitter, Runnable cleanup) { activeStreams.incrementAndGet(); AtomicBoolean completed = new AtomicBoolean(); Runnable finish = () -> { if (completed.compareAndSet(false, true)) { + cleanup.run(); activeStreams.decrementAndGet(); synchronized (streamCompletionMonitor) { streamCompletionMonitor.notifyAll(); @@ -331,6 +350,12 @@ private T track(T emitter) { return emitter; } + private static ResponseEntity toResponse(RestHandler.HTTPRestResponse response) { + return ResponseEntity.status(response.getStatusCode()) + .contentType(A2A_JSON) + .body(response.getBody()); + } + private ServerCallContext buildContext(HttpServletRequest req) { Map state = new LinkedHashMap<>(); Map headers = new LinkedHashMap<>(); diff --git a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java index 2207bdc5..a9f57d05 100644 --- a/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java +++ b/spring-boot-starter/src/test/java/dev/openan/workflow/engine/spring/A2AAutoConfigurationTest.java @@ -21,13 +21,41 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.request; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; +import static org.hamcrest.Matchers.containsString; import java.util.concurrent.ExecutorService; +import java.util.concurrent.Flow; import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.auth.TaskOperation; import org.a2aproject.sdk.server.config.A2AConfigProvider; +import org.a2aproject.sdk.server.requesthandlers.RequestHandler; +import org.a2aproject.sdk.spec.AgentCapabilities; +import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.TaskNotFoundError; +import org.a2aproject.sdk.spec.StreamingEventKind; +import org.a2aproject.sdk.transport.rest.handler.RestHandler; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.test.context.runner.WebApplicationContextRunner; +import org.springframework.http.ResponseEntity; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.MvcResult; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -76,11 +104,139 @@ void everyTaskRouteUsesConfiguredA2aPathPrefix() throws Exception { placeholder + "/tasks/{id}:subscribe", subscribeMethod.getAnnotation(PostMapping.class).value()[0]); assertEquals(SseEmitter.class, subscribeMethod.getReturnType()); + assertEquals( + "text/event-stream", subscribeMethod.getAnnotation(PostMapping.class).produces()[0]); var streamMethod = A2AController.class.getMethod( "streamMessage", jakarta.servlet.http.HttpServletRequest.class, String.class); + // SseEmitter (not ResponseEntity): a wildcard ResponseEntity return is not picked up by + // Spring's emitter handler and the stream response degrades to a converter error. assertEquals(SseEmitter.class, streamMethod.getReturnType()); + assertEquals("text/event-stream", streamMethod.getAnnotation(PostMapping.class).produces()[0]); + } + + @Test + void completedSseReleasesPublisherAndSdkEventConsumer() throws Exception { + RestHandler restHandler = mock(RestHandler.class); + RequestHandler requestHandler = mock(RequestHandler.class); + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build()); + AtomicReference> subscriber = + new AtomicReference<>(); + AtomicInteger subscriptionCancellations = new AtomicInteger(); + AtomicInteger eventConsumerCancellations = new AtomicInteger(); + when(requestHandler.onSubscribeToTask(any(), any())) + .thenAnswer( + invocation -> { + ServerCallContext context = invocation.getArgument(1); + context.setEventConsumerCancelCallback(eventConsumerCancellations::incrementAndGet); + Flow.Publisher publisher = + receiver -> { + subscriber.set(receiver); + receiver.onSubscribe( + new Flow.Subscription() { + @Override + public void request(long count) {} + + @Override + public void cancel() { + subscriptionCancellations.incrementAndGet(); + } + }); + }; + return publisher; + }); + var controller = new A2AController(restHandler, requestHandler, agentCard); + MockMvc mockMvc = + MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + MvcResult stream = + mockMvc + .perform(post("/a2a/json/tasks/task-1:subscribe").accept("text/event-stream")) + .andExpect(request().asyncStarted()) + .andReturn(); + subscriber.get().onComplete(); + mockMvc.perform(asyncDispatch(stream)).andExpect(status().isOk()); + + assertEquals(1, subscriptionCancellations.get()); + assertEquals(1, eventConsumerCancellations.get()); + assertEquals(0, controller.activeStreamCount()); + } + + @Test + void nonStreamingA2AResponseUsesProtocolMediaType() throws Exception { + RestHandler restHandler = mock(RestHandler.class); + when(restHandler.sendMessage(any(), eq(""), eq("{}"))) + .thenReturn(new RestHandler.HTTPRestResponse(200, "application/json", "{}")); + var controller = + new A2AController(restHandler, mock(RequestHandler.class), mock(AgentCard.class)); + MockMvc mockMvc = + MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + mockMvc + .perform(post("/a2a/json/message:send").content("{}")) + .andExpect(status().isOk()) + .andExpect(content().contentTypeCompatibleWith("application/a2a+json")); + } + + @Test + void taskSubscriptionRejectionReturnsAStandardNonStreamingA2AError() throws Exception { + RestHandler restHandler = mock(RestHandler.class); + RequestHandler requestHandler = mock(RequestHandler.class); + var responseBody = + "{\"error\":{\"code\":404,\"status\":\"NOT_FOUND\"," + + "\"message\":\"Task not found\",\"details\":[]}}"; + when(restHandler.createErrorResponse(any())) + .thenReturn(new RestHandler.HTTPRestResponse(404, "application/a2a+json", responseBody)); + doThrow(new TaskNotFoundError()) + .when(requestHandler) + .authorizeTaskAccess(eq("missing"), any(), eq(TaskOperation.SUBSCRIBE_TO_TASK)); + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build()); + var controller = new A2AController(restHandler, requestHandler, agentCard); + + // Through MockMvc so the exception -> @ExceptionHandler mapping runs exactly as in dispatch. + MockMvc mockMvc = + MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + mockMvc + .perform(post("/a2a/json/tasks/missing:subscribe")) + .andExpect(status().is(404)) + .andExpect(content().contentTypeCompatibleWith("application/a2a+json")) + .andExpect(content().string(containsString("\"status\":\"NOT_FOUND\""))); + } + + @Test + void streamingCapabilityIsCheckedBeforeInvokingBusinessHandler() throws Exception { + RestHandler restHandler = mock(RestHandler.class); + RequestHandler requestHandler = mock(RequestHandler.class); + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(false).build()); + when(restHandler.createErrorResponse(any())) + .thenReturn( + new RestHandler.HTTPRestResponse( + 501, + "application/a2a+json", + "{\"error\":{\"code\":-32004,\"status\":\"UNIMPLEMENTED\"}}")); + var controller = new A2AController(restHandler, requestHandler, agentCard); + + MockMvc mockMvc = + MockMvcBuilders.standaloneSetup(controller) + .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json") + .build(); + + mockMvc + .perform(post("/a2a/json/message:stream").content("{}")) + .andExpect(status().is(501)) + .andExpect(content().contentTypeCompatibleWith("application/a2a+json")); + verifyNoInteractions(requestHandler); } @Test From b152007d7703fa1182ebcd133c813d24f189224f Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Fri, 4 Sep 2026 20:35:53 +0800 Subject: [PATCH 3/4] fix(samples): manage the JDK server event bus lifecycle and E2E errors MainEventBusProcessor.ensureStarted() is an intentional CDI no-op, so manual wiring must start the processor thread explicitly; the previous replacement of the reflective start left the bus dead and every inbound task hung. Start the thread in start(), interrupt and join it in close(), create the SSL context before binding the listen address, and unify the two SSE endpoints' write and cleanup paths including the SDK event-consumer cancel callback. Extend the failure E2E to assert that standard 400/429 envelopes reach the northbound failure with stable error codes and no negotiation or merge dispatch. Signed-off-by: Zhoujie628 --- .../examples/server/JdkHttpA2AServer.java | 371 +++++++++++------- .../demo/SpringSpnDemoFailureE2ETest.java | 53 ++- .../server/EmbeddedA2AServerTest.java | 95 +++++ .../workbench/WorkbenchTaskFlowTest.java | 16 +- 4 files changed, 380 insertions(+), 155 deletions(-) diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java index 68a325bc..e36de612 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/server/JdkHttpA2AServer.java @@ -45,7 +45,9 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLParameters; @@ -54,10 +56,12 @@ import org.a2aproject.sdk.grpc.utils.ProtoUtils; import org.a2aproject.sdk.server.AgentCardCacheMetadata; import org.a2aproject.sdk.server.ServerCallContext; +import org.a2aproject.sdk.server.auth.TaskOperation; import org.a2aproject.sdk.server.agentexecution.AgentExecutor; import org.a2aproject.sdk.server.events.InMemoryQueueManager; import org.a2aproject.sdk.server.events.MainEventBus; import org.a2aproject.sdk.server.events.MainEventBusProcessor; +import org.a2aproject.sdk.server.extensions.A2AExtensions; import org.a2aproject.sdk.server.requesthandlers.DefaultRequestHandler; import org.a2aproject.sdk.server.requesthandlers.RequestHandler; import org.a2aproject.sdk.server.tasks.BasePushNotificationSender; @@ -65,14 +69,20 @@ import org.a2aproject.sdk.server.tasks.InMemoryTaskStore; import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.PushNotificationSender; +import org.a2aproject.sdk.server.version.A2AVersionValidator; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.AgentCard; +import org.a2aproject.sdk.spec.InternalError; +import org.a2aproject.sdk.spec.InvalidRequestError; import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.StreamingEventKind; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; +import org.a2aproject.sdk.spec.TaskIdParams; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; +import org.a2aproject.sdk.spec.UnsupportedOperationError; import org.a2aproject.sdk.transport.rest.handler.RestHandler; +import org.a2aproject.sdk.transport.rest.handler.RestHandler.HTTPRestResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -101,6 +111,8 @@ public class JdkHttpA2AServer implements AutoCloseable { private final ExecutorService agentExecutorService; + private final Thread eventBusThread; + private final Map agentCardMap; private final String agentName; @@ -141,7 +153,8 @@ public JdkHttpA2AServer( t.setDaemon(true); return t; }); - RestHandler restHandler = initServerComponents(agentCard, agentExecutor, typedCard); + ServerComponents components = initServerComponents(agentExecutor, typedCard); + this.eventBusThread = components.eventBusThread(); this.executorService = new ThreadPoolExecutor( THREAD_COUNT, @@ -154,8 +167,8 @@ public JdkHttpA2AServer( t.setDaemon(true); return t; }); - this.server = HttpsServer.create(new InetSocketAddress(host, port), 0); SSLContext sslContext = createSslContext(); + this.server = HttpsServer.create(new InetSocketAddress(host, port), 0); this.server.setHttpsConfigurator( new HttpsConfigurator(sslContext) { @Override @@ -166,13 +179,14 @@ public void configure(HttpsParameters params) { }); this.server.setExecutor(executorService); this.server.createContext( - pathPrefix.isEmpty() ? "/" : pathPrefix, exchange -> handleExchange(exchange, restHandler)); + pathPrefix.isEmpty() ? "/" : pathPrefix, + exchange -> handleExchange(exchange, components, typedCard)); if (hasSecuritySchemes(agentCard)) { this.server.createContext("/rest/plat/smapp/v1/oauth/token", this::handleLogin); log.info("[{}] Auth login endpoint enabled", agentName); } - log.info("[{}] A2A server started on https://{}:{}/", agentName, host, port); + log.info("[{}] A2A server configured on https://{}:{}/", agentName, host, port); } private static String formatSse(long seq, String payload) { @@ -209,28 +223,27 @@ private static StreamResponse toStreamResponse(StreamingEventKind event) { throw new IllegalArgumentException("Unsupported event: " + event); } - private static RequestHandler extractRequestHandler(RestHandler restHandler) { - try { - var f = RestHandler.class.getDeclaredField("requestHandler"); - f.setAccessible(true); - return (RequestHandler) f.get(restHandler); - } catch (ReflectiveOperationException e) { - throw new IllegalStateException("Failed to access request handler", e); - } - } - private static ServerCallContext buildCallContext(HttpExchange exchange) { Map headers = new LinkedHashMap<>(); exchange .getRequestHeaders() .forEach((k, v) -> headers.put(k.toLowerCase(Locale.ROOT), String.join(",", v))); String ext = firstHeader(headers, "A2A-Extensions", "X-A2A-Extensions"); - Set exts = ext.isBlank() ? Set.of() : Set.of(ext.split(",")); + Set exts; + if (ext.isBlank()) { + exts = Set.of(); + } else { + java.util.LinkedHashSet parsed = new java.util.LinkedHashSet<>(); + for (String value : ext.split(",")) { + if (!value.isBlank()) parsed.add(value.strip()); + } + exts = java.util.Collections.unmodifiableSet(parsed); + } return new ServerCallContext( null, Map.of("headers", headers), exts, - firstNullableHeader(headers, "A2A-Protocol-Version", "A2A-Version")); + firstNullableHeader(headers, "A2A-Version", "A2A-Protocol-Version")); } private static String firstHeader(Map h, String a, String b) { @@ -257,6 +270,29 @@ private static void sendJson(HttpExchange exchange, int status, Object data) thr } } + private static void sendRestResponse(HttpExchange exchange, HTTPRestResponse response) + throws IOException { + byte[] bytes = response.getBody().getBytes(StandardCharsets.UTF_8); + exchange + .getResponseHeaders() + .set("Content-Type", "application/a2a+json"); + exchange.sendResponseHeaders(response.getStatusCode(), bytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(bytes); + } + } + + private static void sendA2AError( + HttpExchange exchange, RestHandler restHandler, A2AError error) throws IOException { + var response = restHandler.createErrorResponse(error); + byte[] bytes = response.getBody().getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/a2a+json"); + exchange.sendResponseHeaders(response.getStatusCode(), bytes.length); + try (OutputStream os = exchange.getResponseBody()) { + os.write(bytes); + } + } + private static AgentCard toTypedAgentCard(Map card) { return cardMapper.convertValue(card, AgentCard.class); } @@ -288,16 +324,6 @@ private static String extractPathPrefix(Map agentCard) { return ""; } - private static void startEventBus(MainEventBusProcessor proc) { - try { - var m = MainEventBusProcessor.class.getDeclaredMethod("start"); - m.setAccessible(true); - m.invoke(proc); - } catch (ReflectiveOperationException e) { - throw new IllegalStateException("Failed to start event bus", e); - } - } - /** Parses exactly one encoded task-id path segment and rejects ambiguous task routes. */ static String taskIdFromPath(String path, String operationSuffix) { String prefix = "/tasks/"; @@ -345,15 +371,31 @@ private SSLContext createSslContext() throws IOException { } public void start() { - server.start(); + eventBusThread.start(); + try { + server.start(); + log.info( + "[{}] A2A server started on https://{}:{}/", + agentName, + server.getAddress().getHostString(), + server.getAddress().getPort()); + } catch (RuntimeException error) { + eventBusThread.interrupt(); + throw error; + } } @Override public void close() { server.stop(0); + eventBusThread.interrupt(); agentExecutorService.shutdownNow(); executorService.shutdownNow(); try { + eventBusThread.join(1_000); + if (eventBusThread.isAlive()) { + log.warn("[{}] Event bus processor did not stop within 1 second", agentName); + } agentExecutorService.awaitTermination(1, TimeUnit.SECONDS); executorService.awaitTermination(1, TimeUnit.SECONDS); } catch (InterruptedException e) { @@ -370,8 +412,11 @@ public String getAgentName() { return agentName; } - private RestHandler initServerComponents( - Map agentCard, AgentExecutor agentExecutor, AgentCard typedCard) { + boolean isEventBusProcessorAlive() { + return eventBusThread.isAlive(); + } + + private ServerComponents initServerComponents(AgentExecutor agentExecutor, AgentCard typedCard) { InMemoryTaskStore taskStore = new InMemoryTaskStore(); MainEventBus mainEventBus = new MainEventBus(); InMemoryQueueManager queueManager = new InMemoryQueueManager(taskStore, mainEventBus); @@ -379,7 +424,11 @@ private RestHandler initServerComponents( PushNotificationSender pushSender = new BasePushNotificationSender(pushStore); MainEventBusProcessor eventBusProc = new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager); - startEventBus(eventBusProc); + // Manual wiring: no CDI/Spring container invokes the @PostConstruct start(), and + // ensureStarted() is intentionally empty in the SDK (it only forces CDI proxy resolution). + // Without this thread the event bus never drains and inbound tasks hang forever. + Thread eventBusThread = new Thread(eventBusProc, "MainEventBusProcessor"); + eventBusThread.setDaemon(true); RequestHandler requestHandler = DefaultRequestHandler.builder() .agentExecutor(agentExecutor) @@ -390,11 +439,18 @@ private RestHandler initServerComponents( .executor(agentExecutorService) .eventConsumerExecutor(agentExecutorService) .build(); - return new RestHandler( - typedCard, new AgentCardCacheMetadata(typedCard, null), requestHandler, Runnable::run); + RestHandler restHandler = + new RestHandler( + typedCard, new AgentCardCacheMetadata(typedCard, null), requestHandler, Runnable::run); + return new ServerComponents(restHandler, requestHandler, eventBusThread); } - private void handleExchange(HttpExchange exchange, RestHandler restHandler) throws IOException { + private record ServerComponents( + RestHandler restHandler, RequestHandler requestHandler, Thread eventBusThread) {} + + private void handleExchange( + HttpExchange exchange, ServerComponents components, AgentCard typedCard) throws IOException { + RestHandler restHandler = components.restHandler(); String fullPath = exchange.getRequestURI().getRawPath(); String method = exchange.getRequestMethod(); String path = @@ -410,13 +466,18 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro if ("POST".equalsIgnoreCase(method) && "/message:send".equals(path)) { var resp = restHandler.sendMessage(buildCallContext(exchange), "", readBody(exchange)); - sendJson(exchange, resp.getStatusCode(), resp.getBody()); + sendRestResponse(exchange, resp); return; } if ("POST".equalsIgnoreCase(method) && "/message:stream".equals(path)) { - handleStream(exchange, restHandler, readBody(exchange)); + handleStream( + exchange, + restHandler, + components.requestHandler(), + typedCard, + readBody(exchange)); return; } if ("GET".equalsIgnoreCase(method) && "/tasks".equals(path)) { @@ -432,34 +493,36 @@ private void handleExchange(HttpExchange exchange, RestHandler restHandler) thro integerParam(query, "historyLength"), query.get("statusTimestampAfter"), booleanParam(query, "includeArtifacts")); - sendJson(exchange, resp.getStatusCode(), resp.getBody()); + sendRestResponse(exchange, resp); return; } // Task routes (?? A2A: ??/??/??) String taskId = taskIdFromPath(path, null); if ("GET".equalsIgnoreCase(method) && taskId != null) { var resp = restHandler.getTask(buildCallContext(exchange), "", taskId, null); - sendJson(exchange, resp.getStatusCode(), resp.getBody()); + sendRestResponse(exchange, resp); return; } taskId = taskIdFromPath(path, ":cancel"); if ("POST".equalsIgnoreCase(method) && taskId != null) { var resp = restHandler.cancelTask(buildCallContext(exchange), "", readBody(exchange), taskId); - sendJson(exchange, resp.getStatusCode(), resp.getBody()); + sendRestResponse(exchange, resp); return; } taskId = taskIdFromPath(path, ":subscribe"); if ("POST".equalsIgnoreCase(method) && taskId != null) { - handleSubscribe(exchange, restHandler, taskId); + handleSubscribe(exchange, restHandler, components.requestHandler(), typedCard, taskId); return; } exchange.sendResponseHeaders(404, -1); + } catch (A2AError e) { + sendA2AError(exchange, restHandler, e); } catch (IllegalArgumentException e) { - sendJson(exchange, 400, Map.of("error", Map.of("code", 400, "message", e.getMessage()))); + sendA2AError(exchange, restHandler, new InvalidRequestError("Invalid A2A request")); } catch (Exception e) { - log.error("[{}] Handler error: {}", agentName, e.getMessage(), e); - sendJson(exchange, 500, Map.of("error", Map.of("code", 500, "message", e.getMessage()))); + log.error("[{}] Handler error: {}", agentName, e.getClass().getSimpleName(), e); + sendA2AError(exchange, restHandler, new InternalError("A2A request handling failed")); } finally { exchange.close(); } @@ -523,120 +586,156 @@ private void handleLogin(HttpExchange exchange) throws IOException { } @SuppressWarnings("unchecked") - private void handleSubscribe(HttpExchange exchange, RestHandler restHandler, String taskId) + private void handleSubscribe( + HttpExchange exchange, + RestHandler restHandler, + RequestHandler requestHandler, + AgentCard typedCard, + String taskId) throws IOException { try { - var resp = restHandler.subscribeToTask(buildCallContext(exchange), "", taskId); - if (resp instanceof RestHandler.HTTPRestStreamingResponse streaming) { - exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); - exchange.sendResponseHeaders(200, 0); - AtomicLong seq = new AtomicLong(0); - CountDownLatch done = new CountDownLatch(1); - OutputStream os = exchange.getResponseBody(); - streaming - .getPublisher() - .subscribe( - new Flow.Subscriber() { - private Flow.Subscription sub; - - public void onSubscribe(Flow.Subscription s) { - sub = s; - s.request(1); - } - - public void onNext(String item) { - try { - String sse = - String.format(Locale.ROOT, "id:%d%n", seq.incrementAndGet()) - + "data:" - + item - + "\n\n"; - os.write(sse.getBytes(StandardCharsets.UTF_8)); - os.flush(); - sub.request(1); - } catch (IOException e) { - sub.cancel(); - done.countDown(); - } - } - - public void onError(Throwable t) { - done.countDown(); - } - - public void onComplete() { - done.countDown(); - } - }); - try { - done.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - os.close(); - } - } else { - sendJson(exchange, resp.getStatusCode(), resp.getBody()); - } + requireStreamingSupport(typedCard); + ServerCallContext context = buildCallContext(exchange); + requestHandler.authorizeTaskAccess(taskId, context, TaskOperation.SUBSCRIBE_TO_TASK); + Flow.Publisher publisher = + requestHandler.onSubscribeToTask( + TaskIdParams.builder().id(taskId).tenant("").build(), context); + writeStream(exchange, publisher, context); } catch (A2AError e) { - sendJson(exchange, 500, Map.of("error", e.getMessage())); + sendA2AError(exchange, restHandler, e); } } @SuppressWarnings("unchecked") - private void handleStream(HttpExchange exchange, RestHandler restHandler, String requestBody) + private void handleStream( + HttpExchange exchange, + RestHandler restHandler, + RequestHandler requestHandler, + AgentCard typedCard, + String requestBody) throws IOException { try { - RequestHandler inner = extractRequestHandler(restHandler); + requireStreamingSupport(typedCard); + ServerCallContext context = buildCallContext(exchange); + A2AVersionValidator.validateProtocolVersion(typedCard, context); + A2AExtensions.validateRequiredExtensions(typedCard, context); SendMessageRequest.Builder builder = SendMessageRequest.newBuilder(); JsonFormat.parser().merge(requestBody, builder); var request = ProtoUtils.FromProto.messageSendParams(builder.build()); + requestHandler.authorizeTaskAccess( + request.message().taskId(), context, TaskOperation.MESSAGE_SEND_STREAM); Flow.Publisher publisher = - inner.onMessageSendStream(request, buildCallContext(exchange)); - exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); - exchange.sendResponseHeaders(200, 0); - AtomicLong seq = new AtomicLong(0); - CountDownLatch done = new CountDownLatch(1); - OutputStream os = exchange.getResponseBody(); - publisher.subscribe( - new Flow.Subscriber<>() { - private Flow.Subscription sub; - - public void onSubscribe(Flow.Subscription s) { - sub = s; - s.request(1); - } + requestHandler.onMessageSendStream(request, context); + writeStream(exchange, publisher, context); + } catch (A2AError e) { + sendA2AError(exchange, restHandler, e); + } catch (com.google.protobuf.InvalidProtocolBufferException e) { + sendA2AError(exchange, restHandler, new InvalidRequestError("Invalid A2A request body")); + } + } - public void onNext(StreamingEventKind item) { - try { - String payload = JsonFormat.printer().print(toStreamResponse(item)); - os.write( - formatSse(seq.incrementAndGet(), payload).getBytes(StandardCharsets.UTF_8)); - os.flush(); - sub.request(1); - } catch (IOException e) { - sub.cancel(); - done.countDown(); - } - } + private static void requireStreamingSupport(AgentCard agentCard) + throws UnsupportedOperationError { + if (!agentCard.capabilities().streaming()) { + throw new UnsupportedOperationError(null, "Streaming is not supported by the agent", null); + } + } - public void onError(Throwable t) { - done.countDown(); - } + private static void writeStream( + HttpExchange exchange, + Flow.Publisher publisher, + ServerCallContext context) + throws IOException { + exchange.getResponseHeaders().set("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, 0); + AtomicLong seq = new AtomicLong(0); + CountDownLatch done = new CountDownLatch(1); + OutputStream output = exchange.getResponseBody(); + StreamCancellation cancellation = new StreamCancellation(context); + publisher.subscribe( + new Flow.Subscriber<>() { + @Override + public void onSubscribe(Flow.Subscription subscription) { + cancellation.register(subscription); + } - public void onComplete() { + @Override + public void onNext(StreamingEventKind item) { + try { + String payload = JsonFormat.printer().print(toStreamResponse(item)); + output.write( + formatSse(seq.incrementAndGet(), payload).getBytes(StandardCharsets.UTF_8)); + output.flush(); + cancellation.requestNext(); + } catch (IOException error) { + cancellation.release(); done.countDown(); } - }); + } + + @Override + public void onError(Throwable error) { + cancellation.release(); + done.countDown(); + } + + @Override + public void onComplete() { + cancellation.release(); + done.countDown(); + } + }); + try { + done.await(); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + } finally { + cancellation.release(); + output.close(); + } + } + + private static final class StreamCancellation { + private final ServerCallContext context; + private final AtomicReference subscription = new AtomicReference<>(); + private final AtomicBoolean released = new AtomicBoolean(); + + private StreamCancellation(ServerCallContext context) { + this.context = context; + } + + private void register(Flow.Subscription candidate) { + if (!subscription.compareAndSet(null, candidate) || released.get()) { + candidate.cancel(); + } else { + candidate.request(1); + } + } + + private void requestNext() { + Flow.Subscription active = subscription.get(); + if (active != null && !released.get()) { + active.request(1); + } + } + + private void release() { + if (!released.compareAndSet(false, true)) { + return; + } + Flow.Subscription active = subscription.get(); + if (active != null) { + try { + active.cancel(); + } catch (RuntimeException error) { + log.warn("[SSE] Failed to cancel publisher subscription: {}", error.getMessage()); + } + } try { - done.await(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } finally { - os.close(); + context.invokeEventConsumerCancelCallback(); + } catch (RuntimeException error) { + log.warn("[SSE] Failed to notify SDK event consumer cancellation: {}", error.getMessage()); } - } catch (A2AError e) { - sendJson(exchange, 500, Map.of("error", e.getMessage())); } } } diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java index 04edc60c..a9369b4c 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/demo/SpringSpnDemoFailureE2ETest.java @@ -53,7 +53,8 @@ protected String[] arguments(String city1, String city2) throws Exception { @ParameterizedTest @ValueSource(ints = {400, 429}) @org.junit.jupiter.api.Timeout(60) - void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status) throws Exception { + void standardA2AErrorReachesNorthboundFailureWithoutMergeOrResubmission(int status) + throws Exception { var firstCalls = new AtomicInteger(); var secondCalls = new AtomicInteger(); HttpServer city1 = omc(status, firstCalls); @@ -70,7 +71,7 @@ void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status) var error = assertThrows(IllegalStateException.class, () -> new SpringSpnDemo().run(arguments(first, second))); assertTrue(error.getMessage().contains("TASK_STATE_FAILED"), error.getMessage()); - assertTrue(error.getMessage().contains("remote.problem." + status), error.getMessage()); + assertTrue(error.getMessage().contains(errorCode(status)), error.getMessage()); assertTrue(error.getMessage().contains(reason(status)), error.getMessage()); assertTrue(error.getMessage().contains("diagnosis_city2")); assertEquals(1, firstCalls.get()); @@ -79,11 +80,14 @@ void remoteProblemReachesNorthboundFailureWithoutMergeOrResubmission(int status) assertTrue(evidence.contains("OPERATION_FAILED extension=Authorization-T")); assertTrue(evidence.contains("OPERATION_FAILED extension=Notification-T")); assertTrue(evidence.contains("[Executor] TASK_FAILED executionId=")); - assertTrue(evidence.contains("errorCode=remote.problem." + status)); + assertTrue(evidence.contains("errorCode=" + errorCode(status))); assertTrue(evidence.contains("skippedSteps=[merge_analysis]")); assertTrue(evidence.contains("[Orchestrator] WORKFLOW_FINISH success=false, history=2")); - assertTrue(evidence.contains("=== SSE data (JSON display; not wire text) ===")); - assertTrue(evidence.contains("\"status\": " + status)); + assertTrue(evidence.contains("\"error\"")); + // The projected errorDetails keep the observed numeric HTTP status; the envelope's + // "status" field is the canonical string (INVALID_ARGUMENT), not the number. + assertTrue(evidence.contains("\"httpStatus\":" + status)); + assertTrue(evidence.contains("\"status\":\"" + canonicalStatus(status) + "\"")); assertTrue(evidence.contains("[SpringWorkbench] TASK_FAILED")); assertTrue(evidence.contains("[Demo] SHUTDOWN_DONE success=false")); assertFalse(evidence.contains("[ERROR] null")); @@ -128,6 +132,30 @@ private static String reason(int status) { : "Current active tasks have reached the maximum limit 10. Please try again later."; } + private static String errorCode(int status) { + return status == 400 ? "a2a.invalid_params" : "a2a.active_task_limit_exceeded"; + } + + private static String canonicalStatus(int status) { + return status == 400 ? "INVALID_ARGUMENT" : "RESOURCE_EXHAUSTED"; + } + + private static String errorResponse(int status, String message) throws java.io.IOException { + String canonical = status == 400 ? "INVALID_ARGUMENT" + : status == 429 ? "RESOURCE_EXHAUSTED" : "UNAVAILABLE"; + String errorReason = status == 400 ? "INVALID_PARAMS" + : status == 429 ? "ACTIVE_TASK_LIMIT_EXCEEDED" : "OPERATION_REJECTED"; + String domain = status == 400 ? "a2a-protocol.org" : "fixture.invalid"; + return JSON.writeValueAsString(Map.of("error", Map.of( + "code", status, + "status", canonical, + "message", message, + "details", List.of(Map.of( + "@type", "type.googleapis.com/google.rpc.ErrorInfo", + "reason", errorReason, + "domain", domain))))); + } + private static HttpServer omc(int status, AtomicInteger calls) throws Exception { HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); server.createContext("/", exchange -> { @@ -150,11 +178,10 @@ private static HttpServer omc(int status, AtomicInteger calls) throws Exception String response; if (!task) { // Deliberate pre-position failures must not prevent the subsequent workflow tasks. - response = JSON.writeValueAsString(Map.of("status", 503, "detail", "Independent operation rejected")); + response = errorResponse(503, "Independent operation rejected"); } else if (status != 0) { calls.incrementAndGet(); - response = JSON.writeValueAsString(Map.of( - "status", status, "detail", reason(status), "type", "", "timestamp", "2026-08-31T09:07:35Z")); + response = errorResponse(status, reason(status)); } else { calls.incrementAndGet(); response = JSON.writeValueAsString(Map.of("task", Map.of( @@ -162,9 +189,13 @@ private static HttpServer omc(int status, AtomicInteger calls) throws Exception "status", Map.of("state", "TASK_STATE_COMPLETED"), "artifacts", List.of(Map.of("artifactId", "diagnosis", "parts", List.of(Map.of("text", "City2 diagnosis"))))))); } - byte[] bytes = ("data: " + response + "\n\n").getBytes(StandardCharsets.UTF_8); - exchange.getResponseHeaders().set("Content-Type", "text/event-stream;charset=UTF-8"); - exchange.sendResponseHeaders(200, 0); + boolean failed = !task || status != 0; + byte[] bytes = (failed ? response : "data: " + response + "\n\n") + .getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set( + "Content-Type", failed ? "application/a2a+json" : "text/event-stream;charset=UTF-8"); + if (status == 429) exchange.getResponseHeaders().set("Retry-After", "5"); + exchange.sendResponseHeaders(failed ? (!task ? 503 : status) : 200, bytes.length); try (var output = exchange.getResponseBody()) { output.write(bytes); } finally { exchange.close(); } }); 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 1fd25f9c..0ee41902 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 @@ -200,6 +200,101 @@ void testGetAgentCard() throws Exception { assertEquals(AGENT_NAME, card.get("name")); } + @Test + void closeStopsTheManuallyManagedEventBusProcessor() { + assertTrue(server.isEventBusProcessorAlive()); + + server.close(); + + assertFalse(server.isEventBusProcessorAlive()); + server = null; + } + + @Test + void nonStreamingTaskErrorUsesA2AJsonMediaType() throws Exception { + HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build(); + HttpRequest req = + HttpRequest.newBuilder() + .uri(URI.create("https://127.0.0.1:" + port + "/tasks/missing")) + .GET() + .build(); + + HttpResponse resp = http.send(req, HttpResponse.BodyHandlers.ofString()); + + assertEquals(404, resp.statusCode()); + assertEquals( + "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow()); + } + + @Test + void malformedStreamingRequestReturnsAStandardA2AErrorBeforeSseStarts() throws Exception { + HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build(); + HttpRequest req = + HttpRequest.newBuilder() + .uri(URI.create("https://127.0.0.1:" + port + "/message:stream")) + .header("Content-Type", "application/a2a+json") + .header("A2A-Version", "1.0") + .POST(HttpRequest.BodyPublishers.ofString("{")) + .build(); + HttpResponse resp = http.send(req, HttpResponse.BodyHandlers.ofString()); + assertEquals(400, resp.statusCode()); + assertEquals( + "application/a2a+json", + resp.headers().firstValue("Content-Type").orElseThrow()); + var root = mapper.readTree(resp.body()); + assertEquals(400, root.path("error").path("code").asInt()); + assertEquals("INVALID_ARGUMENT", root.path("error").path("status").asText()); + assertTrue(root.path("error").path("details").isArray()); + assertFalse(root.path("error").path("details").isEmpty()); + assertEquals( + "type.googleapis.com/google.rpc.ErrorInfo", + root.path("error").path("details").get(0).path("@type").asText()); + } + + @Test + void unsupportedProtocolVersionReturnsAStandardA2AErrorBeforeSseStarts() throws Exception { + HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build(); + HttpRequest req = + HttpRequest.newBuilder() + .uri(URI.create("https://127.0.0.1:" + port + "/message:stream")) + .header("Content-Type", "application/a2a+json") + .header("A2A-Version", "2.0") + .POST(HttpRequest.BodyPublishers.ofString("{}")) + .build(); + + HttpResponse resp = http.send(req, HttpResponse.BodyHandlers.ofString()); + + assertEquals(400, resp.statusCode()); + assertEquals( + "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow()); + var root = mapper.readTree(resp.body()); + assertEquals("UNIMPLEMENTED", root.path("error").path("status").asText()); + assertEquals( + "VERSION_NOT_SUPPORTED", + root.path("error").path("details").get(0).path("reason").asText()); + } + + @Test + void missingTaskSubscriptionReturnsAStandardA2AErrorBeforeSseStarts() throws Exception { + HttpClient http = HttpClient.newBuilder().sslContext(createTrustAllSslContext()).build(); + HttpRequest req = + HttpRequest.newBuilder() + .uri(URI.create("https://127.0.0.1:" + port + "/tasks/missing:subscribe")) + .POST(HttpRequest.BodyPublishers.noBody()) + .build(); + + HttpResponse resp = http.send(req, HttpResponse.BodyHandlers.ofString()); + + assertEquals(404, resp.statusCode()); + assertEquals( + "application/a2a+json", resp.headers().firstValue("Content-Type").orElseThrow()); + var root = mapper.readTree(resp.body()); + assertEquals(404, root.path("error").path("code").asInt()); + assertEquals("NOT_FOUND", root.path("error").path("status").asText()); + assertEquals( + "TASK_NOT_FOUND", root.path("error").path("details").get(0).path("reason").asText()); + } + @Test void testSendMessage() throws Exception { SendMessageResult result = diff --git a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java index 6fe016ff..379a95c3 100644 --- a/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java +++ b/samples/src/test/java/dev/openan/workflow/engine/examples/workbench/WorkbenchTaskFlowTest.java @@ -51,17 +51,17 @@ void failureSummaryPreservesEachCityReasonWithoutPassingPartialOutputsAsDiagnosi "status", "failed", "errorCode", - "remote.problem.429", + "a2a.active_task_limit_exceeded", "error", "OMC capacity reached", "errorDetails", Map.of( - "status", + "httpStatus", 429, - "detail", - "capacity reached", - "type", - "", + "status", + "RESOURCE_EXHAUSTED", + "reason", + "ACTIVE_TASK_LIMIT_EXCEEDED", "accessSession", "private-token")), Map.of( @@ -76,8 +76,8 @@ void failureSummaryPreservesEachCityReasonWithoutPassingPartialOutputsAsDiagnosi .build(); String summary = WorkbenchOrchestrator.buildResultText(result); assertTrue(summary.contains("city1")); - assertTrue(summary.contains("remote.problem.429")); - assertTrue(summary.contains("capacity reached")); + assertTrue(summary.contains("a2a.active_task_limit_exceeded")); + assertTrue(summary.contains("RESOURCE_EXHAUSTED")); assertTrue(summary.contains("city2")); org.junit.jupiter.api.Assertions.assertFalse(summary.contains("private-token")); org.junit.jupiter.api.Assertions.assertFalse(summary.contains("partial diagnosis")); From 34adc4ee631e0cfcb90167a614b4c7f99e5be1dd Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Fri, 4 Sep 2026 20:35:59 +0800 Subject: [PATCH 4/4] docs: describe the standard A2A error contract Split pre-task-creation failures (non-2xx HTTP plus the google.rpc.Status envelope) from post-creation task failures (HTTP 200 with TASK_STATE_FAILED carrying business evidence). Document the zero-event fast-fail, the rule against fabricating HTTP statuses the transport cannot observe, the six server endpoints with their media types, and the stream cleanup semantics. Signed-off-by: Zhoujie628 --- CHANGELOG.md | 1 + docs/en/API_REFERENCE.md | 20 +++++++++---- docs/en/INTEGRATION_GUIDE.md | 54 ++++++++++++++++++++---------------- docs/zh/API_REFERENCE.md | 17 ++++++++---- docs/zh/INTEGRATION_GUIDE.md | 43 ++++++++++++++-------------- 5 files changed, 79 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f038705a..9645f226 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ - Propagate workflow cancellation and close execution-owned resources exactly once. - Anonymize orchestration access tokens and bound registry response deadlines. - Preserve multi-value results from arbitrary terminal workflow nodes. +- Map pre-task failures from the standard A2A error envelope while keeping post-creation failures in task state. - Publish a compiled host integration example and require sample regression tests. - Keep documentation host-neutral; sample class and AgentCard identifiers remain unchanged. diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index 64823920..f3db2ca0 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -256,7 +256,7 @@ runtime, diagnostic adapter, or explicit lifecycle requires them. | `ProtocolResponses` | A2A event/result assembly helpers | | `ClientEventMapper` | Stable event projection for callbacks and diagnostics | | `WireLog` | Correlation context and protocol-observation facade | -| `RemoteProblemException` | Structured remote problem response (`status`, `title`, `detail`, `type`, `timestamp`) | +| `RemoteA2AErrorException` | Standard A2A HTTP error projection (`code`, `status`, `message`, typed details and safe headers) | --- @@ -589,11 +589,11 @@ not add private negotiation-state keys to the wire message. ## Error Handling -- Top-level remote problem responses retain their fields in `RemoteProblemException`; - `findIn(Throwable)` finds it through wrappers. Workflow failures use `remote.problem.`, - with details in history / TASK_RESPONSE rather than business outputs. +- Standard top-level A2A error envelopes retain their fields in `RemoteA2AErrorException`; + `findIn(Throwable)` also projects wrapped SDK errors. Workflow failures use `a2a.` or + `a2a.http.`, with details in history / TASK_RESPONSE rather than business outputs. Executor node events include executionId; task response events and history include the logical taskId. - See [integration guidance](INTEGRATION_GUIDE.md#14-remote-problem-responses) for propagation and logging. + See [integration guidance](INTEGRATION_GUIDE.md#14-a2a-errors-and-task-failures) for propagation and logging. - Missing/null/exceptional/timed-out callbacks fail explicitly. Host SDK errors may be mapped to BusinessFailure with safe code/details. @@ -655,7 +655,7 @@ Auto-configures the following beans (all `@ConditionalOnMissingBean`, so you can | `eventBusProcessor` | `MainEventBusProcessor` | Event bus processor | | `requestHandler` | `RequestHandler` | Default request handler | | `restHandler` | `RestHandler` | REST protocol handler | -| `a2aController` | `A2AController` | Spring MVC controller (`message:send` + `message:stream`) | +| `a2aController` | `A2AController` | Spring MVC controller for message and task endpoints | ### A2AController @@ -663,6 +663,14 @@ Spring MVC controller that exposes A2A REST endpoints: - `POST {path-prefix}/message:send` — blocking send - `POST {path-prefix}/message:stream` — SSE streaming +- `GET {path-prefix}/tasks/{id}` — query a task +- `GET {path-prefix}/tasks` — list tasks with optional filters +- `POST {path-prefix}/tasks/{id}:cancel` — cancel a task +- `POST {path-prefix}/tasks/{id}:subscribe` — subscribe to task updates over SSE + +HTTP+JSON responses use `application/a2a+json`; successful streaming responses use `text/event-stream`. When a stream +closes, the controller cancels its publisher subscription and notifies the SDK event consumer so that server-side +polling does not remain active. ### Usage diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index dac010d9..58fea3cb 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -549,28 +549,34 @@ Cleanup is enabled and fail-fast by default so stale tasks cannot silently lead the demo may cancel active tasks created by another installation; use an isolated identity or disable cleanup after providing an equivalent ownership-aware policy. -## 14. Remote problem responses - -A successful HTTP envelope does not guarantee a successful task: SSE data can contain a top-level -problem object such as `{"status":429,"detail":"Active task limit reached","type":""}`. -Top-level numeric status 400–599 with title/detail becomes `RemoteProblemException`, preserving -status, title, detail, type and timestamp. Both synchronous and streaming responses are checked. -Nested business data is not classified as a protocol error. -Detection is limited to bodies actually delivered by the SDK. A non-2xx HTTP response may be rejected -first as an HTTP/authentication error by the SDK or adapter, without exposing its problem body; -such failures retain the generic failure path rather than fabricating errorDetails. - -The error fails the call without fabricating a successful task, invoking onNegotiation, or automatically -resubmitting it. For 400, inspect business inputs; for 429, inspect dispatched-agent capacity and active tasks or -subscriptions before deciding whether to retry. Negotiation still requires a valid Negotiation-T Propose. -Use `RemoteProblemException.findIn(error)` to inspect the cause chain (null means no problem found); -workflow execution follows its existing failure path. -Task results and execution history / TASK_RESPONSE events expose `remote.problem.400`, -`remote.problem.429`, etc. as errorCode, the reason as error, the five fields above as errorDetails, -and empty outputs. Known credential fields use the protocol logger's redaction rules; -unrecognized SDK exceptions still expose only the exception type. -Independent authorization/subscription failures do not determine the workflow result. -Pretty logging does not modify the received problem payload. +## 14. A2A errors and task failures + +Keep failures before and after task creation separate. A request rejected before a task is created uses a +non-2xx HTTP response with the standard A2A `google.rpc.Status` JSON envelope: + +```json +{"error":{"code":400,"status":"INVALID_ARGUMENT","message":"A required parameter is missing","details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo","reason":"INVALID_PARAMS","domain":"a2a-protocol.org","metadata":{"field":"port"}}]}} +``` + +`RemoteA2AErrorException` preserves the observed HTTP status, envelope code/status/message, typed details, +ErrorInfo reason/domain and safe response headers. `findIn(Throwable)` also projects typed A2A Java SDK +errors. The direct transport checks ordinary responses and defensively detects a top-level error envelope if +an SDK exposes a pre-stream rejection as SSE data. Nested `error` objects inside a task, message or artifact +remain business content. + +If an SSE call closes before delivering any A2A event, the transport fails immediately instead of waiting for the +workflow send timeout. The underlying SDK does not expose the HTTP status of an empty streaming response, so this case +is reported as a transport/protocol failure rather than a fabricated HTTP error code. + +Workflow history and TASK_RESPONSE expose a stable code such as `a2a.invalid_params`; unknown HTTP errors +fall back to `a2a.http.`. `errorDetails` retains protocol facts and `retryAfter` when observed. Such an +error creates no successful output, does not invoke onNegotiation and is not retried automatically. + +After a task has been created, a business execution failure is not an HTTP protocol error. The agent returns +HTTP 200 with a Task or status update in `TASK_STATE_FAILED`; the TaskStatus message, task metadata and +artifacts carry the available failure evidence. The engine marks the workflow task failed, keeps that evidence +in `receivedMessages`, and does not interpret an extension-specific result schema. Independent authorization +and subscription failures remain independent of workflow outcome. For an `ALL_SUCCESS` step, one failed task does not retroactively cancel a peer call that was already dispatched: the peer result or timeout is collected before the workflow returns `success=false`. Downstream steps are not invoked; @@ -584,9 +590,9 @@ The host agent decides how to expose a failed execution to its caller. It must n success artifact. A persistent host owns notification subscriptions independently of workflow results. Logging responsibilities: PROTOCOL captures observed traffic, correlated by requestId rather than -adjacent log entries; REMOTE_PROBLEM reports transport rejection; TASK_FAILED identifies the execution, +adjacent log entries; A2A_ERROR reports a protocol rejection; TASK_FAILED identifies the execution, step, logical task, agent, code and reason; WORKFLOW_STOPPED lists unexecuted steps. -Sample TASK_RESPONSE logs bridge contextId and executionId. Recognized remote problems use WARN summaries, +Sample TASK_RESPONSE logs bridge contextId and executionId. Recognized A2A errors use WARN summaries, while unexpected exceptions retain ERROR stack traces. Known credentials and Bearer/Basic values are redacted. Logging configuration, pretty display and observer callbacks do not determine task outcomes. diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index 749c06cc..96b0c505 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -244,7 +244,7 @@ Map normalized = AgentCardNormalizer.normalize(rawMap); | `ProtocolResponses` | A2A 事件与结果组装辅助方法 | | `ClientEventMapper` | 面向回调和诊断的稳定事件投影 | | `WireLog` | 关联上下文与协议观测门面 | -| `RemoteProblemException` | 结构化远程问题响应（`status`、`title`、`detail`、`type`、`timestamp`） | +| `RemoteA2AErrorException` | 标准 A2A HTTP 错误投影（`code`、`status`、`message`、类型化详情及安全响应头） | --- @@ -560,10 +560,10 @@ getReceivedMessages() 是保留层级的响应来源，getOutputs() 为便利投 ## 错误处理 -- 顶层远端 problem 错误通过 `RemoteProblemException` 保留；可用 `findIn(Throwable)` 查找包装异常中的原因。 - 工作流任务失败信息为 `remote.problem.`，详情进入 history / TASK_RESPONSE，不作为业务 outputs。 +- 顶层标准 A2A 错误信封通过 `RemoteA2AErrorException` 保留；`findIn(Throwable)` 也会投影包装的 SDK 错误。 + 工作流任务失败信息为 `a2a.` 或 `a2a.http.`，详情进入 history / TASK_RESPONSE，不作为业务 outputs。 引擎节点事件携带 executionId，任务结果事件和 history 携带逻辑 taskId。 - 失败传播、并行步骤失败规则和日志职责见 [集成指南](INTEGRATION_GUIDE.md#14-远端错误响应)。 + 失败传播、并行步骤失败规则和日志职责见 [集成指南](INTEGRATION_GUIDE.md#14-a2a-错误与任务失败)。 - 回调异常／空值／超时明确失败；SDK 内容错误由宿主转换为 BusinessFailure，必要时保留安全的 code/details。 - maxNegotiationExchanges 耗尽或本地 Stop 只结束本地交互，不自动发送 Abort。业务 Send(Abort) 不算任务成功。 @@ -621,7 +621,7 @@ a2at: | `eventBusProcessor` | `MainEventBusProcessor` | 事件总线处理器 | | `requestHandler` | `RequestHandler` | 默认请求处理器 | | `restHandler` | `RestHandler` | REST 协议处理器 | -| `a2aController` | `A2AController` | Spring MVC 控制器（`message:send` + `message:stream`） | +| `a2aController` | `A2AController` | 消息和任务端点的 Spring MVC 控制器 | ### A2AController @@ -629,6 +629,13 @@ a2at: - `POST {path-prefix}/message:send` — 阻塞式发送 - `POST {path-prefix}/message:stream` — SSE 流式 +- `GET {path-prefix}/tasks/{id}` — 查询任务 +- `GET {path-prefix}/tasks` — 按可选条件分页查询任务 +- `POST {path-prefix}/tasks/{id}:cancel` — 取消任务 +- `POST {path-prefix}/tasks/{id}:subscribe` — 通过 SSE 订阅任务更新 + +HTTP+JSON 响应使用 `application/a2a+json`，成功建立的流式响应使用 `text/event-stream`。连接结束时，控制器会取消 +Publisher 订阅并通知 SDK 停止事件消费，避免服务端轮询继续占用资源。 ### 用法 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index fdfe9768..c0eceb10 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -514,30 +514,31 @@ Demo 在打开独立协议通道和启动工作流之前，先查询每个被调 `A2A_TASK_CLEANUP_MAX_TASKS` 配置。任务查询受认证身份权限约束；若多个实例共用同一身份，Demo 可能取消其他实例创建的活跃任务。 应使用隔离身份，或在提供等价的任务归属清理策略后关闭该功能。 -## 14. 远端错误响应 +## 14. A2A 错误与任务失败 -响应头成功不代表任务成功。HTTP 200 的 SSE data 也可能返回顶层错误对象，例如： +任务创建前和创建后的失败必须分开处理。请求在任务创建前被拒绝时，返回非 2xx HTTP 状态和标准 +A2A `google.rpc.Status` JSON 错误信封： ```json -{"status":429,"detail":"Current active tasks have reached the maximum limit 10.","type":""} +{"error":{"code":400,"status":"INVALID_ARGUMENT","message":"缺少必填参数","details":[{"@type":"type.googleapis.com/google.rpc.ErrorInfo","reason":"INVALID_PARAMS","domain":"a2a-protocol.org","metadata":{"field":"port"}}]}} ``` -通信层识别顶层数字 status 为 400–599 且携带 title/detail 的错误对象， -以 `RemoteProblemException` 保留 status、title、detail、type、timestamp。 -同步响应和流式响应均检查；普通任务内容或 artifact 内嵌的同名字段不被当成协议错误。 -该识别以 SDK 实际交付的正文为限：真实 HTTP 非 2xx 可能先被 SDK/适配器作为 HTTP 或认证错误拒绝， -不保证可取得 problem 正文；没有该正文时仍走原有通用失败路径，不伪造 errorDetails。 -错误不生成成功结果、不触发 onNegotiation，也不自动重新提交请求。400 应核对业务参数； -429 应检查被调度智能体的容量、活跃任务和订阅，再由业务决定是否重试。 -只有合法 Negotiation-T Propose 才进入协商回调。 - -直接调用客户端时可用 `RemoteProblemException.findIn(error)` 检查 cause 链（没有则返回 null）； -工作流执行时沿现有失败路径报告错误： -任务结果及执行 history / TASK_RESPONSE 事件中的 errorCode 为 `remote.problem.400`、 -`remote.problem.429` 等，error 包含原因，errorDetails 包含上述五个字段，outputs 为空。 -错误内容中的已知凭据字段按协议日志的规则脱敏；未识别的 SDK 异常仍只暴露异常类型。 -预置授权、通知订阅的错误仍由其独立生命周期处理，不决定工作流执行结果。 -协议 pretty 日志仅用于展示，不改写原始收到的错误对象。 +`RemoteA2AErrorException` 保留实际观察到的 HTTP 状态、信封 code/status/message、类型化 details、 +ErrorInfo reason/domain 及安全响应头；`findIn(Throwable)` 也会投影 A2A Java SDK 的类型化错误。 +直连传输检查普通响应，并防御性识别被 SDK 作为 SSE data 暴露的建流前顶层错误信封。 +Task、Message 或 Artifact 内嵌的 `error` 对象仍属于业务内容，不会被误判为协议错误。 + +若 SSE 调用未产生任何 A2A 事件便结束，传输会立即失败，不再等待工作流发送超时。底层 SDK 不暴露空流响应的 HTTP +状态，因此该场景报告为传输/协议失败，不会虚构 HTTP 错误码。 + +工作流 history 和 TASK_RESPONSE 使用 `a2a.invalid_params` 等稳定错误码；未知 HTTP 错误回退为 +`a2a.http.`。`errorDetails` 保留协议事实及实际取得的 `retryAfter`。这类错误不生成成功输出、 +不触发 onNegotiation，也不自动重试。 + +任务创建成功后的业务执行失败不是 HTTP 协议错误。被调度智能体返回 HTTP 200，并通过 Task 或 +StatusUpdate 的 `TASK_STATE_FAILED` 表示失败；TaskStatus 消息、任务 metadata 和 artifact 携带失败证据。 +引擎将工作流任务标记为失败，把证据保留在 `receivedMessages`，但不解释扩展协议的业务结果 schema。 +独立授权和通知订阅的失败仍按各自生命周期处理，不决定工作流结果。 对于 `ALL_SUCCESS` 步骤，一个任务失败不会追溯取消已下发的并行任务；引擎在另一任务返回或超时后报告 `success=false`。后续步骤不执行，成功的并行结果保留在 history，不伪造整体成功结果。引擎不自动重试、排队或部分成功汇总；`ANY_SUCCESS` 节点仍遵循声明的任一成功语义。 @@ -547,9 +548,9 @@ Demo 在打开独立协议通道和启动工作流之前，先查询每个被调 宿主智能体决定如何向调用方暴露失败，但不得把失败执行转换为成功 artifact。常驻宿主以独立生命周期管理通知订阅。 **日志职责**：PROTOCOL 记录已观察到的请求/响应，按 requestId 关联并发报文； -REMOTE_PROBLEM 记录通信层识别的远端拒绝，TASK_FAILED 记录执行 ID、节点、逻辑任务 ID、 +A2A_ERROR 记录协议拒绝，TASK_FAILED 记录执行 ID、节点、逻辑任务 ID、 智能体、错误码和原因，WORKFLOW_STOPPED 列出未执行节点。示例 TASK_RESPONSE 日志同时包含 -contextId 与 executionId，供跨层定位。已识别的远端问题使用 WARN 摘要， +contextId 与 executionId，供跨层定位。已识别的 A2A 错误使用 WARN 摘要， 未知异常保留 ERROR 堆栈；错误详情中的已知凭据及 Bearer/Basic 值脱敏。 日志开关、pretty 展示和观察回调不会改变任务结果。