From cdd08a95d114651eac1ea1d4ea5d98dce212273f Mon Sep 17 00:00:00 2001 From: XianReallyHot-ZZH <741788769@qq.com> Date: Sun, 15 Nov 2026 02:21:03 +0800 Subject: [PATCH] fix(a2a): propagate transport, protocol and remote task failures as structured errors (issue #213) --- .../agent/orchestration/a2a/A2aException.java | 46 ++ .../a2a/A2aNodeActionWithConfig.java | 350 +++++++++--- .../a2a/A2aProtocolException.java | 43 ++ .../a2a/A2aRemoteTaskException.java | 47 ++ .../a2a/A2aTransportException.java | 43 ++ .../a2a/A2aNodeActionWithConfigTests.java | 38 +- .../A2aStructuredFailurePropagationTest.java | 519 ++++++++++++++++++ 7 files changed, 1004 insertions(+), 82 deletions(-) create mode 100644 argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aException.java create mode 100644 argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aProtocolException.java create mode 100644 argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aRemoteTaskException.java create mode 100644 argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aTransportException.java create mode 100644 argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aStructuredFailurePropagationTest.java diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aException.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aException.java new file mode 100644 index 000000000..9f1d72272 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aException.java @@ -0,0 +1,46 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.a2a; + +/** + * Base type of the structured A2A client failures. Transport, protocol and remote task + * failures are propagated as these exceptions (issue #213) so callers can distinguish a + * failed remote call from a normal completion whose text merely mentions an error. + * + *

The hierarchy extends {@link IllegalStateException} because that is the exception + * type the A2A client historically threw for these conditions; existing + * {@code catch (IllegalStateException)} callers keep working unchanged.

+ */ +public class A2aException extends IllegalStateException { + + /** + * Constructs a new A2A failure with the specified error message. + * @param message the detail message + */ + protected A2aException(String message) { + super(message); + } + + /** + * Constructs a new A2A failure with the specified error message and cause. + * @param message the detail message + * @param cause the cause + */ + protected A2aException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfig.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfig.java index c154b76da..d9b2f2376 100644 --- a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfig.java +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfig.java @@ -32,6 +32,7 @@ import org.apache.hc.client5.http.impl.classic.HttpClients; import org.apache.hc.core5.http.ContentType; import org.apache.hc.core5.http.HttpEntity; +import org.apache.hc.core5.http.ParseException; import org.apache.hc.core5.http.io.entity.EntityUtils; import org.apache.hc.core5.http.io.entity.StringEntity; @@ -44,8 +45,10 @@ import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStream; import java.io.InputStreamReader; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -173,9 +176,25 @@ public Map apply(OverAllState state, RunnableConfig config) thro String requestPayload = buildSendMessageRequest(state, subGraphRunnableConfig, context); String resultText = sendRequest(requestPayload, context); try (AgentTraceSpan receiveSpan = startSpan(AgentTraceOperation.A2A_RECEIVE, context).role("consumer")) { - Map resultMap = autoDetectAndParseResponse(resultText); - Map result = (Map) resultMap.get("result"); - String responseText = extractResponseText(result); + final String responseText; + try { + Map resultMap = autoDetectAndParseResponse(resultText); + final Map result = resultPayload(resultMap); + final A2aRemoteTaskException remoteFailure = result != null ? remoteTaskFailure(result) : null; + if (remoteFailure != null) { + throw remoteFailure; + } + responseText = extractResponseText(result); + } + catch (A2aException ex) { + receiveSpan.error(ex); + throw ex; + } + catch (IllegalArgumentException | IllegalStateException ex) { + A2aProtocolException protocol = new A2aProtocolException("A2A response cannot be parsed", ex); + receiveSpan.error(protocol); + throw protocol; + } receiveSpan.success(); return buildFinalResult(this.outputKeyToParent, responseText); } @@ -233,9 +252,11 @@ public String subGraphId() { /** * Stream the remote agent response. The HTTP exchange starts on subscription and runs * on a bounded elastic worker. Each non-empty text chunk is emitted as a - * {@link StreamingOutput}; transport and protocol failures are emitted as error text - * chunks. The stream always ends with a done response carrying the accumulated text - * for the parent state, unless the subscriber cancels first. + * {@link StreamingOutput}; transport, protocol and remote task failures propagate as + * {@link A2aException} terminal errors (issue #213), so a failed remote call is never + * mistaken for a normal completion. On success the stream ends with a done response + * carrying the accumulated text for the parent state, unless the subscriber cancels + * first. */ private Flux> createStreamingFlux(OverAllState state, RunnableConfig config, AgentTraceContext context) throws Exception { @@ -246,11 +267,11 @@ private Flux> createStreamingFlux(OverAllState state, final StringBuilder accumulated = new StringBuilder(); String baseUrl = resolveAgentBaseUrl(this.agentCard); if (baseUrl == null || baseUrl.isBlank()) { - emit(sink, buildStreamingOutput("Error: AgentCard.url is empty", state)); - complete(sink, outputKey, accumulated); + fail(sink, new A2aTransportException("AgentCard.url is empty")); return; } + A2aException failure = null; try (CloseableHttpClient httpClient = HttpClients.createDefault()) { HttpPost post = new HttpPost(baseUrl); post.setHeader("Content-Type", "application/json"); @@ -273,17 +294,12 @@ private Flux> createStreamingFlux(OverAllState state, httpClient.execute(post, response -> { int statusCode = response.getCode(); if (statusCode != 200) { - StreamingOutput errorOutput = buildStreamingOutput("HTTP request failed, status: " - + statusCode, state); - emit(sink, errorOutput); - return null; + throw new A2aTransportException("HTTP request failed, status: " + statusCode); } HttpEntity entity = response.getEntity(); if (entity == null) { - StreamingOutput errorOutput = buildStreamingOutput("Empty HTTP entity", state); - emit(sink, errorOutput); - return null; + throw new A2aTransportException("Empty HTTP entity"); } String contentType = entity.getContentType() != null ? entity.getContentType() : ""; @@ -293,52 +309,10 @@ private Flux> createStreamingFlux(OverAllState state, .role("consumer"); try { if (isEventStream) { - try (BufferedReader reader = new BufferedReader( - new InputStreamReader(entity.getContent(), StandardCharsets.UTF_8))) { - String line; - while (!sink.isCancelled() && (line = reader.readLine()) != null) { - String trimmed = line.trim(); - if (!trimmed.startsWith("data:")) { - continue; - } - String jsonContent = trimmed.substring(5).trim(); - if ("[DONE]".equals(jsonContent)) { - break; - } - try { - Map parsed = parseJsonMap(jsonContent); - Map result = (Map) parsed.get("result"); - if (result != null) { - String text = extractResponseText(result); - if (text != null && !text.isEmpty()) { - accumulated.append(text); - emit(sink, buildStreamingOutput(text, state)); - } - } - } - catch (Exception ignore) { - } - } - } + consumeEventStream(entity.getContent(), state, sink, accumulated); } else { - // Non-SSE: read the full body and emit a single output - String body = EntityUtils.toString(entity, "UTF-8"); - try { - Map resultMap = parseJsonMap(body); - Map result = (Map) resultMap.get("result"); - String text = extractResponseText(result); - if (text != null && !text.isEmpty()) { - accumulated.append(text); - emit(sink, buildStreamingOutput(text, state)); - } - } - catch (Exception ex) { - if (!sink.isCancelled()) { - receiveSpan.error(ex); - } - emit(sink, buildStreamingOutput("Error: " + ex.getMessage(), state)); - } + consumeSingleBodyResponse(entity, state, sink, accumulated); } if (!sink.isCancelled()) { receiveSpan.success(); @@ -369,15 +343,261 @@ private Flux> createStreamingFlux(OverAllState state, sendSpan.close(); } } + catch (A2aException e) { + failure = e; + } catch (Exception e) { - emit(sink, buildStreamingOutput("Error: " + e.getMessage(), state)); + failure = asA2aFailure(e); } - finally { - complete(sink, outputKey, accumulated); + + if (failure != null) { + // The failure is the single terminal state: text chunks already delivered + // keep their identity and no done response follows an error. + fail(sink, failure); + return; } + complete(sink, outputKey, accumulated); }).subscribeOn(Schedulers.boundedElastic()); } + /** + * Consume an A2A {@code text/event-stream} response. Events follow the SSE wire format: + * comment lines and fields other than {@code data} are ignorable extensions, the + * {@code data} lines of one event are joined with newlines, and a blank line dispatches + * the accumulated event. Reading stops at the {@code [DONE]} sentinel or a completed + * status-update; an event stream that ends without any terminal state is a protocol + * failure (issue #213). + */ + private void consumeEventStream(InputStream content, OverAllState state, + FluxSink> sink, StringBuilder accumulated) throws IOException { + try (BufferedReader reader = new BufferedReader(new InputStreamReader(content, StandardCharsets.UTF_8))) { + final List dataLines = new ArrayList<>(2); + String line; + while (!sink.isCancelled() && (line = reader.readLine()) != null) { + if (line.isEmpty()) { + // a blank line dispatches the accumulated event per the SSE wire format + if (!dataLines.isEmpty() && dispatchSseEvent(String.join("\n", dataLines), state, sink, + accumulated)) { + return; + } + dataLines.clear(); + continue; + } + if (line.startsWith(":")) { + // comments and keep-alive lines carry no data + continue; + } + if (line.startsWith("data:")) { + dataLines.add(line.substring(5).trim()); + } + // other field names (event:, id:, retry:) are ignorable extensions + } + if (!sink.isCancelled() && !dataLines.isEmpty()) { + // the peer may close without a trailing blank line; the pending event + // still counts + if (dispatchSseEvent(String.join("\n", dataLines), state, sink, accumulated)) { + return; + } + } + if (!sink.isCancelled()) { + throw new A2aProtocolException("A2A event stream ended without a terminal state"); + } + } + } + + /** + * Handle one assembled SSE data event. + * @return whether the event terminates the remote task: the {@code [DONE]} sentinel or + * a completed status-update + */ + private boolean dispatchSseEvent(String data, OverAllState state, FluxSink> sink, + StringBuilder accumulated) { + if ("[DONE]".equals(data)) { + return true; + } + final Map parsed; + try { + parsed = parseJsonMap(data); + } + catch (IllegalArgumentException ex) { + throw new A2aProtocolException("Corrupt A2A SSE frame, not valid JSON: " + abbreviate(data), ex); + } + final Map result = resultPayload(parsed); + if (result == null) { + // valid JSON carrying no result object: an ignorable extension event + return false; + } + final A2aRemoteTaskException remoteFailure = remoteTaskFailure(result); + if (remoteFailure != null) { + throw remoteFailure; + } + if (isCompletedStatus(result)) { + return true; + } + final String text; + try { + text = extractResponseText(result); + } + catch (Exception ex) { + // a recognizable frame without extractable text is skipped, mirroring the + // historical behavior for events this client does not consume + return false; + } + if (text != null && !text.isEmpty()) { + accumulated.append(text); + emit(sink, buildStreamingOutput(text, state)); + } + return false; + } + + /** + * The result object carried by one parsed JSON-RPC payload, or null when the payload has + * none. A JSON-RPC error object translates to a protocol failure here, so both payload + * shapes (SSE frames and single bodies) classify errors identically. + */ + private Map resultPayload(Map parsed) { + if (parsed.containsKey("error")) { + throw jsonRpcFailure(parsed.get("error")); + } + return parsed.get("result") instanceof Map rawResult ? (Map) rawResult : null; + } + + /** + * Consume a non-SSE response body: one JSON-RPC payload carrying the whole remote + * result. Parse failures, JSON-RPC errors and remote task failures propagate as + * structured errors; a single body is by itself the terminal state (issue #213). + */ + private void consumeSingleBodyResponse(HttpEntity entity, OverAllState state, + FluxSink> sink, StringBuilder accumulated) throws IOException, ParseException { + final String body = EntityUtils.toString(entity, "UTF-8"); + final Map parsed; + try { + parsed = parseJsonMap(body); + } + catch (IllegalArgumentException ex) { + throw new A2aProtocolException("A2A response body is not valid JSON", ex); + } + final Map result = resultPayload(parsed); + if (result == null) { + throw new A2aProtocolException("A2A response without a result object"); + } + final A2aRemoteTaskException remoteFailure = remoteTaskFailure(result); + if (remoteFailure != null) { + throw remoteFailure; + } + final String text; + try { + text = extractResponseText(result); + } + catch (Exception ex) { + throw new A2aProtocolException("A2A response without usable content", ex); + } + if (text != null && !text.isEmpty()) { + accumulated.append(text); + emit(sink, buildStreamingOutput(text, state)); + } + } + + /** + * Deliver the structured failure as the stream's single terminal state. Text chunks + * already emitted keep their identity; a cancelled subscription gets neither the error + * nor a done response. + */ + private static void fail(FluxSink> sink, A2aException failure) { + if (!sink.isCancelled()) { + sink.error(failure); + } + } + + /** + * Map an exception thrown by the streaming exchange to the structured A2A failure. The + * Apache client may wrap handler exceptions, so an A2A failure is recovered from the + * cause chain; anything else is a transport failure. + */ + private static A2aException asA2aFailure(Throwable error) { + Throwable current = error; + for (int depth = 0; current != null && depth < 8; depth++) { + if (current instanceof A2aException a2a) { + return a2a; + } + current = current.getCause(); + } + return new A2aTransportException("A2A streaming call failed: " + error.getMessage(), error); + } + + /** + * Translate a JSON-RPC error object into a protocol failure carrying the remote code + * and message. + */ + private static A2aProtocolException jsonRpcFailure(Object errorValue) { + final String detail; + if (errorValue instanceof Map error && error.get("message") != null) { + detail = "code=" + error.get("code") + ", message=" + error.get("message"); + } + else { + detail = String.valueOf(errorValue); + } + return new A2aProtocolException("A2A remote protocol error: " + detail); + } + + /** + * Detect a remote task termination that must surface as a structured failure instead of + * a normal completion (issue #213). Both shapes carry the outcome in a + * {@code status.state} field: status-update events on a stream and the Task object a + * single-body response returns. The optional status message carries the remote reason. + * @return the failure to propagate, or null when the payload is not a task failure + */ + private static A2aRemoteTaskException remoteTaskFailure(Map result) { + final String state = remoteTaskState(result); + if (!"failed".equals(state) && !"canceled".equals(state)) { + return null; + } + final String reason = result.get("status") instanceof Map status ? statusMessageText(status) : ""; + return new A2aRemoteTaskException(state, + "Remote A2A task " + state + (reason.isBlank() ? "" : ": " + reason)); + } + + /** + * Best-effort text of an A2A status message: the concatenated text parts, if any. + */ + private static String statusMessageText(Map status) { + if (!(status.get("message") instanceof Map message) + || !(message.get("parts") instanceof List parts)) { + return ""; + } + final StringBuilder text = new StringBuilder(); + for (Object part : parts) { + if (part instanceof Map partMap && partMap.get("text") instanceof String partText) { + text.append(partText); + } + } + return text.toString(); + } + + /** + * The remote task state reported by a status-update event or a Task object, or null for + * payloads carrying no task status. + */ + private static String remoteTaskState(Map result) { + return result.get("status") instanceof Map status && status.get("state") instanceof String state + ? state : null; + } + + /** + * Whether the payload reports the final completed state that terminates a successful + * remote task. + */ + private static boolean isCompletedStatus(Map result) { + return "completed".equals(remoteTaskState(result)); + } + + /** + * Shorten an offending SSE frame for an error message. + */ + private static String abbreviate(String data) { + return data.length() <= 120 ? data : data.substring(0, 120) + "..."; + } + private static void emit(FluxSink> sink, NodeOutput output) { if (!sink.isCancelled()) { sink.next(GraphResponse.of(output)); @@ -762,7 +982,7 @@ else if (!shareState || (shareState && state.value("messages").isEmpty())) { private String sendMessageToServer(AgentCardWrapper agentCard, String requestPayload) throws Exception { String baseUrl = resolveAgentBaseUrl(agentCard); if (baseUrl == null || baseUrl.isBlank()) { - throw new IllegalStateException("AgentCard.url is empty"); + throw new A2aTransportException("AgentCard.url is empty"); } try (CloseableHttpClient httpClient = HttpClients.createDefault()) { @@ -773,11 +993,11 @@ private String sendMessageToServer(AgentCardWrapper agentCard, String requestPay return httpClient.execute(post, response -> { int statusCode = response.getCode(); if (statusCode != 200) { - throw new IllegalStateException("HTTP request failed, status: " + statusCode); + throw new A2aTransportException("HTTP request failed, status: " + statusCode); } HttpEntity entity = response.getEntity(); if (entity == null) { - throw new IllegalStateException("Empty HTTP entity"); + throw new A2aTransportException("Empty HTTP entity"); } return EntityUtils.toString(entity, "UTF-8"); }); diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aProtocolException.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aProtocolException.java new file mode 100644 index 000000000..89f104cbf --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aProtocolException.java @@ -0,0 +1,43 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.a2a; + +/** + * The remote peer violated the A2A protocol: a response frame was not valid JSON, a + * JSON-RPC error object was returned, a response carried no usable result, or an event + * stream ended without a terminal state. The remote task state is unknown; any text + * already delivered before the violation remains valid partial output. + */ +public class A2aProtocolException extends A2aException { + + /** + * Constructs a new protocol failure with the specified error message. + * @param message the detail message + */ + public A2aProtocolException(String message) { + super(message); + } + + /** + * Constructs a new protocol failure with the specified error message and cause. + * @param message the detail message + * @param cause the cause + */ + public A2aProtocolException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aRemoteTaskException.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aRemoteTaskException.java new file mode 100644 index 000000000..4fb82aae4 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aRemoteTaskException.java @@ -0,0 +1,47 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.a2a; + +/** + * The remote A2A task terminated abnormally: it reported a {@code failed} or + * {@code canceled} status-update. This is a definitive remote outcome, not a local + * transport or parsing problem; the optional status message text reported by the remote + * peer is carried in the exception message. + */ +public class A2aRemoteTaskException extends A2aException { + + /** The A2A task state reported by the remote peer, {@code failed} or {@code canceled}. */ + private final String taskState; + + /** + * Constructs a new remote task failure. + * @param taskState the A2A task state reported by the remote peer + * @param message the detail message, including any remote reason text + */ + public A2aRemoteTaskException(String taskState, String message) { + super(message); + this.taskState = taskState; + } + + /** + * Return the A2A task state reported by the remote peer. + * @return the task state, {@code failed} or {@code canceled} + */ + public String taskState() { + return this.taskState; + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aTransportException.java b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aTransportException.java new file mode 100644 index 000000000..d73064287 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/main/java/io/github/agentic/ai/agent/orchestration/a2a/A2aTransportException.java @@ -0,0 +1,43 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.a2a; + +/** + * An A2A call failed at the transport layer: the HTTP exchange returned a non-success + * status, carried no entity, was interrupted, or could not be started at all (for example + * a missing agent card URL). The remote agent was never reached or never answered; no + * statement can be made about its task state. + */ +public class A2aTransportException extends A2aException { + + /** + * Constructs a new transport failure with the specified error message. + * @param message the detail message + */ + public A2aTransportException(String message) { + super(message); + } + + /** + * Constructs a new transport failure with the specified error message and cause. + * @param message the detail message + * @param cause the cause + */ + public A2aTransportException(String message, Throwable cause) { + super(message, cause); + } + +} diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfigTests.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfigTests.java index 412438e37..13ead7a02 100644 --- a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfigTests.java +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aNodeActionWithConfigTests.java @@ -31,6 +31,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.ValueSource; import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; import java.io.ByteArrayOutputStream; import java.io.PrintStream; @@ -87,15 +88,15 @@ void streamingEmitsEventStreamChunksAndAccumulatedResult() throws Exception { } @Test - void streamingReportsHttpFailureAndStillCompletes() throws Exception { + void streamingHttpFailureFailsWithStructuredTransportError() throws Exception { HttpServer server = startServer(500, "application/json", "{}"); try { - List> responses = stream(baseUrl(server)); - - assertEquals(2, responses.size()); - assertEquals("HTTP request failed, status: 500", streamingText(responses.get(0))); - assertTrue(responses.get(1).isDone()); - assertEquals("", ((Map) responses.get(1).resultValue().orElseThrow()).get("reply")); + StepVerifier.create(streamingFlux(baseUrl(server))) + .expectErrorSatisfies(error -> { + assertInstanceOf(A2aTransportException.class, error); + assertEquals("HTTP request failed, status: 500", error.getMessage()); + }) + .verify(Duration.ofSeconds(5)); } finally { server.stop(0); @@ -103,12 +104,11 @@ void streamingReportsHttpFailureAndStillCompletes() throws Exception { } @Test - void streamingWithoutAgentUrlCompletesInsteadOfWaitingForever() throws Exception { - List> responses = stream(null); - - assertEquals(2, responses.size()); - assertEquals("Error: AgentCard.url is empty", streamingText(responses.get(0))); - assertTrue(responses.get(1).isDone()); + void streamingWithoutAgentUrlFailsWithStructuredTransportError() throws Exception { + StepVerifier.create(streamingFlux(null)) + .expectErrorSatisfies( + error -> assertInstanceOf(A2aTransportException.class, error)) + .verify(Duration.ofSeconds(5)); } @Test @@ -171,14 +171,18 @@ void forwardsUserIdFromConfiguredMetadataKey() throws Exception { @SuppressWarnings("unchecked") private static List> stream(String baseUrl) throws Exception { + List> responses = streamingFlux(baseUrl).collectList().block(Duration.ofSeconds(5)); + assertNotNull(responses); + return responses; + } + + @SuppressWarnings("unchecked") + private static Flux> streamingFlux(String baseUrl) throws Exception { A2aNodeActionWithConfig node = new A2aNodeActionWithConfig(createAgentCardWrapper(baseUrl), "remote-agent", false, "reply", "instruction", true); Map result = node.apply(new OverAllState(), RunnableConfig.builder().threadId("stream-test").build()); - Flux> flux = (Flux>) result.get("reply"); - List> responses = flux.collectList().block(Duration.ofSeconds(5)); - assertNotNull(responses); - return responses; + return (Flux>) result.get("reply"); } private static String streamingText(GraphResponse response) { diff --git a/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aStructuredFailurePropagationTest.java b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aStructuredFailurePropagationTest.java new file mode 100644 index 000000000..b2815f7d0 --- /dev/null +++ b/argi-orchestration/argi-agent-framework/src/test/java/io/github/agentic/ai/agent/orchestration/a2a/A2aStructuredFailurePropagationTest.java @@ -0,0 +1,519 @@ +/* + * Copyright 2024-2026 the original author or authors. + * + * 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 + * + * https://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 io.github.agentic.ai.agent.orchestration.a2a; + +import io.github.agentic.ai.workflow.graph.CompileConfig; +import io.github.agentic.ai.workflow.graph.CompiledGraph; +import io.github.agentic.ai.workflow.graph.GraphResponse; +import io.github.agentic.ai.workflow.graph.NodeOutput; +import io.github.agentic.ai.workflow.graph.OverAllState; +import io.github.agentic.ai.workflow.graph.RunnableConfig; +import io.github.agentic.ai.workflow.graph.StateGraph; +import io.github.agentic.ai.workflow.graph.action.AsyncNodeActionWithConfig; +import io.github.agentic.ai.workflow.graph.streaming.StreamingOutput; + +import io.a2a.spec.AgentCard; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import com.sun.net.httpserver.HttpServer; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * Issue #213 (OPT-22): A2A transport and parsing failures must propagate as structured + * errors instead of an ordinary text chunk followed by a normal completion. Each failure + * class carries its own exception type, an error path emits at most one terminal state, + * text already delivered before the failure keeps its identity, and normal remote text + * that merely starts with {@code Error:} is never treated as a control signal. + */ +class A2aStructuredFailurePropagationTest { + + private static final Duration TIMEOUT = Duration.ofSeconds(5); + + // ==================== Streaming: transport failures ==================== + + @Test + void httpFailureFailsTheStreamWithTransportErrorInsteadOfCompleting() throws Exception { + HttpServer server = startServer(500, "application/json", "{}"); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aTransportException.class); + assertThat(error.getMessage()).contains("500"); + // legacy callers catching the historic exception type keep working + assertThat(error).isInstanceOf(IllegalStateException.class); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void missingAgentUrlFailsTheStreamWithTransportError() throws Exception { + StepVerifier.create(streamingFlux(null)) + .expectErrorSatisfies(error -> assertThat(error).isInstanceOf(A2aTransportException.class) + .hasMessageContaining("AgentCard.url is empty")) + .verify(TIMEOUT); + } + + // ==================== Streaming: protocol failures ==================== + + @Test + void corruptSseFrameFailsTheStreamAfterTheChunksAlreadyDelivered() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + "data: " + artifactFrame("partial") + "\n\ndata: {not valid json\n\n"); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("partial")) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aProtocolException.class); + assertThat(error.getMessage()).contains("not valid JSON"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void jsonRpcErrorFrameFailsTheStreamWithProtocolError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + "data: {\"jsonrpc\":\"2.0\",\"id\":\"1\",\"error\":{\"code\":-32601,\"message\":\"Method not found\"}}\n\n"); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aProtocolException.class); + assertThat(error.getMessage()).contains("Method not found").contains("-32601"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void streamEndingWithoutAnyTerminalStateFailsWithProtocolError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + ("data: " + artifactFrame("orphan") + "\n\n")); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("orphan")) + .expectErrorSatisfies( + error -> assertThat(error).isInstanceOf(A2aProtocolException.class) + .hasMessageContaining("terminal")) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void nonSseBodyThatIsNotJsonFailsTheStreamWithProtocolError() throws Exception { + HttpServer server = startServer(200, "application/json", "not-json-at-all"); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies( + error -> assertThat(error).isInstanceOf(A2aProtocolException.class)) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + // ==================== Streaming: remote task failures ==================== + + @Test + void remoteFailedStatusFailsTheStreamWithRemoteTaskError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + ("data: " + statusFrame("failed", "tool crashed") + "\n\n")); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aRemoteTaskException.class); + assertThat(((A2aRemoteTaskException) error).taskState()).isEqualTo("failed"); + assertThat(error.getMessage()).contains("tool crashed"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void remoteCanceledStatusFailsTheStreamWithRemoteTaskError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + ("data: " + statusFrame("canceled", null) + "\n\n")); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies( + error -> assertThat(((A2aRemoteTaskException) error).taskState()).isEqualTo("canceled")) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void remoteFailedSingleBodyResponseFailsTheStreamWithRemoteTaskError() throws Exception { + HttpServer server = startServer(200, "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"1\"," + taskResult("failed", "late rejection") + "}"); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aRemoteTaskException.class); + assertThat(error.getMessage()).contains("late rejection"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void remoteFailedTaskObjectIsNotMisclassifiedAsAProtocolError() throws Exception { + // A single-body response returns a Task object (kind:"task"), not a status-update: + // the failed state must still classify as a remote task failure. + HttpServer server = startServer(200, "application/json", "{\"jsonrpc\":\"2.0\",\"id\":\"1\"," + + taskResult("failed", "task object exploded") + "}"); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aRemoteTaskException.class); + assertThat(((A2aRemoteTaskException) error).taskState()).isEqualTo("failed"); + assertThat(error.getMessage()).contains("task object exploded"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void completedTaskObjectWithArtifactsDeliversOrdinaryText() throws Exception { + HttpServer server = startServer(200, "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"kind\":\"task\"," + + "\"status\":{\"state\":\"completed\"}," + + "\"artifacts\":[{\"parts\":[{\"text\":\"via artifacts\"}]}]}}"); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("via artifacts")) + .assertNext(GraphResponse::isDone) + .verifyComplete(); + } + finally { + server.stop(0); + } + } + + // ==================== Streaming: what must NOT be treated as failure ==================== + + @Test + void remoteTextStartingWithErrorIsDeliveredAsOrdinaryContent() throws Exception { + HttpServer server = startServer(200, "text/event-stream", + ("data: " + artifactFrame("Error: this is merely the remote answer") + "\n\ndata: [DONE]\n\n")); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)) + .isEqualTo("Error: this is merely the remote answer")) + .assertNext(GraphResponse::isDone) + .verifyComplete(); + } + finally { + server.stop(0); + } + } + + @Test + void completedStatusUpdateTerminatesTheStreamWithoutDoneSentinelOrError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", ("data: %s\n\ndata: %s\n\n") + .formatted(artifactFrame("done via status"), statusFrame("completed", null))); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("done via status")) + .assertNext(GraphResponse::isDone) + .verifyComplete(); + } + finally { + server.stop(0); + } + } + + @Test + void sseCommentsUnknownFieldsAndUnknownEventsAreIgnored() throws Exception { + HttpServer server = startServer(200, "text/event-stream", ": keep-alive comment\n\n" + + "event: status\nid: 42\n" + + "data: {\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"kind\":\"some-future-extension\"}}\n\n" + + "data: " + artifactFrame("visible") + "\n\n" + "data: [DONE]\n\n"); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("visible")) + .assertNext(GraphResponse::isDone) + .verifyComplete(); + } + finally { + server.stop(0); + } + } + + @Test + void multiDataLineFrameIsJoinedBeforeParsing() throws Exception { + // A newline inside JSON is ordinary whitespace, so an SSE frame split across two + // data field lines must be reassembled with a newline before parsing. + HttpServer server = startServer(200, "text/event-stream", + "data: {\"jsonrpc\":\"2.0\",\"id\":\"1\",\n" + "data: \"result\":{\"kind\":\"artifact-update\"," + + "\"artifact\":{\"parts\":[{\"text\":\"joined\"}]}}}\n\n" + "data: [DONE]\n\n"); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("joined")) + .assertNext(GraphResponse::isDone) + .verifyComplete(); + } + finally { + server.stop(0); + } + } + + // ==================== Non-streaming: structured failures ==================== + + @Test + void nonStreamingHttpFailureThrowsTransportException() throws Exception { + HttpServer server = startServer(500, "application/json", "{}"); + try { + A2aException error = assertThrows(A2aException.class, () -> nonStreamingCall(server)); + assertThat(error).isInstanceOf(A2aTransportException.class); + assertThat(error).isInstanceOf(IllegalStateException.class); + } + finally { + server.stop(0); + } + } + + @Test + void nonStreamingRemoteFailedThrowsRemoteTaskException() throws Exception { + HttpServer server = startServer(200, "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"1\"," + taskResult("failed", "remote exploded") + "}"); + try { + A2aRemoteTaskException error = assertThrows(A2aRemoteTaskException.class, + () -> nonStreamingCall(server)); + assertThat(error.taskState()).isEqualTo("failed"); + assertThat(error.getMessage()).contains("remote exploded"); + } + finally { + server.stop(0); + } + } + + @Test + void nonStreamingRemoteCanceledThrowsRemoteTaskException() throws Exception { + HttpServer server = startServer(200, "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"1\"," + taskResult("canceled", null) + "}"); + try { + A2aRemoteTaskException error = assertThrows(A2aRemoteTaskException.class, + () -> nonStreamingCall(server)); + assertThat(error.taskState()).isEqualTo("canceled"); + } + finally { + server.stop(0); + } + } + + @Test + void nonStreamingUnparsableBodyThrowsProtocolException() throws Exception { + HttpServer server = startServer(200, "application/json", "garbage"); + try { + assertThrows(A2aProtocolException.class, () -> nonStreamingCall(server)); + } + finally { + server.stop(0); + } + } + + @Test + void nonStreamingRemoteTextStartingWithErrorIsAnOrdinaryResult() throws Exception { + HttpServer server = startServer(200, "application/json", + "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"kind\":\"artifact-update\",\"artifact\":" + + "{\"parts\":[{\"text\":\"Error: just the answer\"}]}}}"); + try { + Map result = nonStreamingCall(server); + assertThat(result.get("reply")).isEqualTo("Error: just the answer"); + } + finally { + server.stop(0); + } + } + + // ==================== Streaming: transport interruption and empty responses ==================== + + @Test + void transportInterruptionAfterPartialTextFailsWithTransportError() throws Exception { + // The server declares more bytes than it sends and closes: the body is truncated + // mid-stream, which is a transport failure, not a clean end of the event stream. + byte[] bytes = ("data: " + artifactFrame("interrupted") + "\n\n").getBytes(StandardCharsets.UTF_8); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + exchange.getResponseHeaders().add("Content-Type", "text/event-stream"); + exchange.sendResponseHeaders(200, bytes.length + 100); + exchange.getResponseBody().write(bytes); + exchange.getResponseBody().flush(); + exchange.close(); + }); + server.start(); + try { + StepVerifier.create(streamingFlux(server)) + .assertNext(response -> assertThat(streamingText(response)).isEqualTo("interrupted")) + .expectErrorSatisfies( + error -> assertThat(error).isInstanceOf(A2aTransportException.class)) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + @Test + void emptyEventStreamFailsWithMissingTerminalError() throws Exception { + HttpServer server = startServer(200, "text/event-stream", ""); + try { + StepVerifier.create(streamingFlux(server)) + .expectErrorSatisfies(error -> { + assertThat(error).isInstanceOf(A2aProtocolException.class); + assertThat(error.getMessage()).contains("terminal"); + }) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + // ==================== Upper layer: the graph surfaces the structured failure ==================== + + @Test + void upperLayerGraphStreamSurfacesTheStructuredFailure() throws Exception { + HttpServer server = startServer(500, "application/json", "{}"); + try { + A2aNodeActionWithConfig node = new A2aNodeActionWithConfig(agentCard(baseUrl(server)), "remote-agent", + false, "reply", "instruction", true); + StateGraph graph = new StateGraph("a2a-upper", HashMap::new); + graph.addNode("remote", AsyncNodeActionWithConfig.node_async(node)); + graph.addEdge(StateGraph.START, "remote"); + graph.addEdge("remote", StateGraph.END); + CompiledGraph compiled = graph.compile(CompileConfig.builder().build()); + + StepVerifier + .create(compiled.stream(Map.of(), RunnableConfig.builder().threadId("upper-layer").build())) + // the __START__ marker output precedes the remote node's failure + .expectNextCount(1) + .expectErrorSatisfies( + error -> assertThat(error).isInstanceOf(A2aTransportException.class)) + .verify(TIMEOUT); + } + finally { + server.stop(0); + } + } + + // ==================== helpers ==================== + + @SuppressWarnings("unchecked") + private static Flux> streamingFlux(HttpServer server) throws Exception { + String url = server == null ? null : baseUrl(server); + A2aNodeActionWithConfig node = new A2aNodeActionWithConfig(agentCard(url), "remote-agent", false, "reply", + "instruction", true); + Map result = node.apply(new OverAllState(), + RunnableConfig.builder().threadId("structured-failure").build()); + return (Flux>) result.get("reply"); + } + + private static Map nonStreamingCall(HttpServer server) throws Exception { + A2aNodeActionWithConfig node = new A2aNodeActionWithConfig(agentCard(baseUrl(server)), "remote-agent", false, + "reply", "instruction", false); + return node.apply(new OverAllState(), RunnableConfig.builder().threadId("structured-failure").build()); + } + + private static String streamingText(GraphResponse response) { + assertTrue(!response.isDone()); + StreamingOutput output = (StreamingOutput) response.getOutput().getNow(null); + return output.text(); + } + + private static String artifactFrame(String text) { + return "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"kind\":\"artifact-update\",\"artifact\":" + + "{\"parts\":[{\"text\":\"" + text + "\"}]}}}"; + } + + private static String statusFrame(String state, String messageText) { + String message = messageText == null ? "" + : ",\"message\":{\"parts\":[{\"text\":\"" + messageText + "\"}]}"; + return "{\"jsonrpc\":\"2.0\",\"id\":\"1\",\"result\":{\"kind\":\"status-update\",\"status\":" + + "{\"state\":\"" + state + "\"" + message + "}}}"; + } + + /** A JSON-RPC envelope whose result is the Task object a single-body response returns. */ + private static String taskResult(String state, String messageText) { + String message = messageText == null ? "" + : ",\"message\":{\"parts\":[{\"text\":\"" + messageText + "\"}]}"; + return "\"result\":{\"id\":\"task-1\",\"contextId\":\"ctx-1\",\"kind\":\"task\"," + + "\"status\":{\"state\":\"" + state + "\"" + message + "}}"; + } + + private static HttpServer startServer(int status, String contentType, String body) throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/", exchange -> { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().add("Content-Type", contentType); + exchange.sendResponseHeaders(status, bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + return server; + } + + private static String baseUrl(HttpServer server) { + return "http://127.0.0.1:" + server.getAddress().getPort() + "/"; + } + + private static AgentCardWrapper agentCard(String url) { + AgentCard agentCard = mock(AgentCard.class); + when(agentCard.name()).thenReturn("test-agent"); + when(agentCard.url()).thenReturn(url); + return new AgentCardWrapper(agentCard); + } + +}