publisher =
- requestHandler.onSubscribeToTask(TaskIdParams.builder().id(taskId).tenant("").build(), ctx);
+ 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());
diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java
index 116152d..df91261 100644
--- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java
+++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2AProperties.java
@@ -40,8 +40,8 @@ public class A2AProperties {
private String pathPrefix = "/a2a/json";
/**
- * Whether to expose slash-style aliases for A2A action endpoints for gateways that reject
- * colons in published paths. Standard colon-style endpoints always remain available.
+ * Whether to expose slash-style aliases for A2A action endpoints for gateways that reject colons
+ * in published paths. Standard colon-style endpoints always remain available.
*/
private boolean slashActionAliasesEnabled = false;
diff --git a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2ASlashActionAliasController.java b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2ASlashActionAliasController.java
index a5ef9a8..56f3b0c 100644
--- a/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2ASlashActionAliasController.java
+++ b/spring-boot-starter/src/main/java/dev/openan/workflow/engine/spring/A2ASlashActionAliasController.java
@@ -30,8 +30,8 @@
/**
* Optional slash-style aliases for A2A action endpoints.
*
- * Every operation delegates to {@link A2AController}; the aliases therefore retain the
- * canonical endpoint's authentication, A2A error envelope, streaming, and cleanup behavior.
+ *
Every operation delegates to {@link A2AController}; the aliases therefore retain the canonical
+ * endpoint's authentication, A2A error envelope, streaming, and cleanup behavior.
*/
@RestController
@ConditionalOnProperty(
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 493953b..f722896 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
@@ -18,6 +18,7 @@
*/
package dev.openan.workflow.engine.spring;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -29,12 +30,11 @@
import static org.mockito.Mockito.verify;
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.request.MockMvcRequestBuilders.post;
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.lang.reflect.Modifier;
import java.util.concurrent.ExecutorService;
@@ -49,13 +49,12 @@
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.spec.TaskNotFoundError;
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;
@@ -169,26 +168,30 @@ public void cancel() {
assertEquals(0, controller.activeStreamCount());
}
-
@org.junit.jupiter.params.ParameterizedTest
@org.junit.jupiter.params.provider.ValueSource(
strings = {"before-subscribe", "after-subscribe", "during-request"})
- void synchronousPublisherFailureReleasesResourcesBeforeMvcDispatch(String phase) throws Exception {
+ void synchronousPublisherFailureReleasesResourcesBeforeMvcDispatch(String phase)
+ throws Exception {
for (String endpoint : new String[] {"/message:stream", "/tasks/task-1:subscribe"}) {
RestHandler restHandler = mock(RestHandler.class);
when(restHandler.createErrorResponse(any()))
- .thenReturn(new RestHandler.HTTPRestResponse(
- 500, "application/a2a+json",
- "{\"error\":{\"code\":500,\"status\":\"INTERNAL\",\"message\":\"Setup failed\"}}"));
+ .thenReturn(
+ new RestHandler.HTTPRestResponse(
+ 500,
+ "application/a2a+json",
+ "{\"error\":{\"code\":500,\"status\":\"INTERNAL\",\"message\":\"Setup failed\"}}"));
RequestHandler requestHandler = mock(RequestHandler.class);
AgentCard agentCard = mock(AgentCard.class);
- when(agentCard.capabilities()).thenReturn(AgentCapabilities.builder().streaming(true).build());
+ when(agentCard.capabilities())
+ .thenReturn(AgentCapabilities.builder().streaming(true).build());
var agentInterface = mock(org.a2aproject.sdk.spec.AgentInterface.class);
when(agentInterface.protocolVersion()).thenReturn("1.0");
when(agentCard.supportedInterfaces()).thenReturn(java.util.List.of(agentInterface));
AtomicInteger subscriptionCancellations = new AtomicInteger();
AtomicInteger eventConsumerCancellations = new AtomicInteger();
- AtomicReference> subscriber = new AtomicReference<>();
+ AtomicReference> subscriber =
+ new AtomicReference<>();
org.mockito.stubbing.Answer> answer =
invocation -> {
ServerCallContext context = invocation.getArgument(1);
@@ -196,19 +199,20 @@ void synchronousPublisherFailureReleasesResourcesBeforeMvcDispatch(String phase)
return receiver -> {
subscriber.set(receiver);
if (!phase.equals("before-subscribe")) {
- receiver.onSubscribe(new Flow.Subscription() {
- @Override
- public void request(long count) {
- if (phase.equals("during-request")) {
- throw new IllegalStateException("Request failed");
- }
- }
-
- @Override
- public void cancel() {
- subscriptionCancellations.incrementAndGet();
- }
- });
+ receiver.onSubscribe(
+ new Flow.Subscription() {
+ @Override
+ public void request(long count) {
+ if (phase.equals("during-request")) {
+ throw new IllegalStateException("Request failed");
+ }
+ }
+
+ @Override
+ public void cancel() {
+ subscriptionCancellations.incrementAndGet();
+ }
+ });
}
throw new IllegalStateException("Subscribe failed");
};
@@ -216,16 +220,20 @@ public void cancel() {
when(requestHandler.onSubscribeToTask(any(), any())).thenAnswer(answer);
when(requestHandler.onMessageSendStream(any(), any())).thenAnswer(answer);
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" + endpoint)
- .header("A2A-Version", "1.0")
- .contentType("application/json")
- .accept("text/event-stream")
- .content("{\"message\":{\"messageId\":\"m1\",\"role\":\"ROLE_USER\","
- + "\"parts\":[{\"text\":\"test\"}]}}"))
+ MockMvc mockMvc =
+ MockMvcBuilders.standaloneSetup(controller)
+ .addPlaceholderValue("a2at.server.path-prefix", "/a2a/json")
+ .build();
+
+ mockMvc
+ .perform(
+ post("/a2a/json" + endpoint)
+ .header("A2A-Version", "1.0")
+ .contentType("application/json")
+ .accept("text/event-stream")
+ .content(
+ "{\"message\":{\"messageId\":\"m1\",\"role\":\"ROLE_USER\","
+ + "\"parts\":[{\"text\":\"test\"}]}}"))
.andExpect(status().isInternalServerError())
.andExpect(request().asyncNotStarted())
.andExpect(content().contentTypeCompatibleWith("application/a2a+json"))
@@ -414,8 +422,7 @@ void slashActionAliasesAreOptIn() {
.withBean(A2AController.class, () -> canonical)
.withPropertyValues("a2at.server.enabled=true");
- runner.run(
- context -> assertFalse(context.containsBean("a2aSlashActionAliasController")));
+ runner.run(context -> assertFalse(context.containsBean("a2aSlashActionAliasController")));
runner
.withPropertyValues("a2at.server.slash-action-aliases-enabled=true")
.run(context -> assertTrue(context.containsBean("a2aSlashActionAliasController")));
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
index c602404..efdf37d 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClient.java
@@ -36,31 +36,74 @@ final class A2AErrorDetectingHttpClient implements A2AHttpClient {
this.delegate = delegate;
}
- @Override public GetBuilder createGet() { return new Get(delegate.createGet()); }
- @Override public PostBuilder createPost() { return new Post(delegate.createPost()); }
- @Override public DeleteBuilder createDelete() { return new Delete(delegate.createDelete()); }
+ @Override
+ public GetBuilder createGet() {
+ return new Get(delegate.createGet());
+ }
+
+ @Override
+ public PostBuilder createPost() {
+ return new Post(delegate.createPost());
+ }
+
+ @Override
+ public DeleteBuilder createDelete() {
+ return new Delete(delegate.createDelete());
+ }
private static A2AHttpResponse check(A2AHttpResponse response) {
- var error = RemoteA2AErrorException.fromResponse(
- response.status(), response.body(), response.headers().toMap());
+ var error =
+ RemoteA2AErrorException.fromResponse(
+ response.status(), response.body(), response.headers().toMap());
if (error != null) throw error;
return response;
}
private abstract static class Wrapped> implements Builder {
final T target;
- Wrapped(T target) { this.target = target; }
+
+ Wrapped(T target) {
+ this.target = target;
+ }
+
abstract T self();
- @Override public T url(String url) { target.url(url); return self(); }
- @Override public T addHeader(String name, String value) { target.addHeader(name, value); return self(); }
- @Override public T addHeaders(Map headers) { target.addHeaders(headers); return self(); }
+
+ @Override
+ public T url(String url) {
+ target.url(url);
+ return self();
+ }
+
+ @Override
+ public T addHeader(String name, String value) {
+ target.addHeader(name, value);
+ return self();
+ }
+
+ @Override
+ public T addHeaders(Map headers) {
+ target.addHeaders(headers);
+ return self();
+ }
}
private static final class Get extends Wrapped implements GetBuilder {
- Get(GetBuilder target) { super(target); }
- @Override GetBuilder self() { return this; }
- @Override public A2AHttpResponse get() throws IOException, InterruptedException { return check(target.get()); }
- @Override public CompletableFuture getAsyncSSE(
+ Get(GetBuilder target) {
+ super(target);
+ }
+
+ @Override
+ GetBuilder self() {
+ return this;
+ }
+
+ @Override
+ public A2AHttpResponse get() throws IOException, InterruptedException {
+ return check(target.get());
+ }
+
+ @Override
+ public CompletableFuture getAsyncSSE(
Consumer messages, Consumer errors, Runnable complete)
throws IOException, InterruptedException {
return stream(target::getAsyncSSE, messages, errors, complete);
@@ -68,11 +111,28 @@ private static final class Get extends Wrapped implements GetBuilder
}
private static final class Post extends Wrapped implements PostBuilder {
- Post(PostBuilder target) { super(target); }
- @Override PostBuilder self() { return this; }
- @Override public PostBuilder body(String body) { target.body(body); return this; }
- @Override public A2AHttpResponse post() throws IOException, InterruptedException { return check(target.post()); }
- @Override public CompletableFuture postAsyncSSE(
+ Post(PostBuilder target) {
+ super(target);
+ }
+
+ @Override
+ PostBuilder self() {
+ return this;
+ }
+
+ @Override
+ public PostBuilder body(String body) {
+ target.body(body);
+ return this;
+ }
+
+ @Override
+ public A2AHttpResponse post() throws IOException, InterruptedException {
+ return check(target.post());
+ }
+
+ @Override
+ public CompletableFuture postAsyncSSE(
Consumer messages, Consumer errors, Runnable complete)
throws IOException, InterruptedException {
return stream(target::postAsyncSSE, messages, errors, complete);
@@ -80,63 +140,93 @@ private static final class Post extends Wrapped implements PostBuil
}
private static final class Delete extends Wrapped implements DeleteBuilder {
- Delete(DeleteBuilder target) { super(target); }
- @Override DeleteBuilder self() { return this; }
- @Override public A2AHttpResponse delete() throws IOException, InterruptedException { return check(target.delete()); }
+ Delete(DeleteBuilder target) {
+ super(target);
+ }
+
+ @Override
+ DeleteBuilder self() {
+ return this;
+ }
+
+ @Override
+ public A2AHttpResponse delete() throws IOException, InterruptedException {
+ return check(target.delete());
+ }
}
private static CompletableFuture stream(
- StreamStarter starter, Consumer messages,
- Consumer errors, Runnable complete) throws IOException, InterruptedException {
+ StreamStarter starter,
+ Consumer messages,
+ Consumer errors,
+ Runnable complete)
+ throws IOException, InterruptedException {
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); }
- catch (RuntimeException observerError) {
- if (observerError != error) error.addSuppressed(observerError);
- }
- finally { result.completeExceptionally(error); }
- };
- result.whenComplete((ignored, error) -> {
- if (error != null) {
- ended.set(true);
- var active = upstream.get();
- if (active != null) active.cancel(true);
- }
- });
- var active = starter.start(event -> {
- if (ended.get()) return;
- 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); }
- }
- });
+ Consumer fail =
+ error -> {
+ if (!ended.compareAndSet(false, true)) return;
+ try {
+ errors.accept(error);
+ } catch (RuntimeException observerError) {
+ if (observerError != error) error.addSuppressed(observerError);
+ } finally {
+ result.completeExceptionally(error);
+ }
+ };
+ result.whenComplete(
+ (ignored, error) -> {
+ if (error != null) {
+ ended.set(true);
+ var active = upstream.get();
+ if (active != null) active.cancel(true);
+ }
+ });
+ var active =
+ starter.start(
+ event -> {
+ if (ended.get()) return;
+ 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);
+ }
+ }
+ });
upstream.set(active);
if (result.isCompletedExceptionally()) active.cancel(true);
- active.whenComplete((ignored, error) -> {
- if (error != null) fail.accept(error);
- });
+ active.whenComplete(
+ (ignored, error) -> {
+ if (error != null) fail.accept(error);
+ });
return result;
}
@FunctionalInterface
private interface StreamStarter {
- CompletableFuture start(Consumer messages, Consumer errors,
- Runnable complete) throws IOException, InterruptedException;
+ CompletableFuture start(
+ Consumer messages, Consumer errors, Runnable complete)
+ throws IOException, InterruptedException;
}
}
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 1c514e0..f939274 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
@@ -542,10 +542,10 @@ public void closeConversation(AgentCard agentCard, String contextId) {
}
/**
- * Long-lived SSE stream for Notification-T subscription. Opens a daemon thread that keeps the
- * SSE response stream open. The eventSink callback processes events in real-time (subscribed ack
- * + later recovery results). The returned future completes on the first event carrying a
- * concrete task state; an artifact alone is application data, not a protocol acknowledgement.
+ * Long-lived SSE stream for Notification-T subscription. Opens a daemon thread that keeps the SSE
+ * response stream open. The eventSink callback processes events in real-time (subscribed ack +
+ * later recovery results). The returned future completes on the first event carrying a concrete
+ * task state; an artifact alone is application data, not a protocol acknowledgement.
*/
public NotificationSubscription openNotificationStream(
AgentCard agentCard,
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHeaderContributor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHeaderContributor.java
index 0ea914a..d304eec 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHeaderContributor.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHeaderContributor.java
@@ -42,9 +42,7 @@ final class CredentialHeaderContributor implements HeaderContributor {
}
private static void addCredentialHeader(
- Map headers,
- Map schemeConfig,
- String credential) {
+ Map headers, Map schemeConfig, String credential) {
String authHeader = (String) schemeConfig.get("auth_header");
if (authHeader != null && !authHeader.isEmpty()) {
String prefix = (String) schemeConfig.getOrDefault("auth_header_prefix", "");
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java
index 7c63ae4..68f9e64 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClient.java
@@ -441,8 +441,8 @@ private CompletableFuture advance(
/**
* Shared negotiation loop for both carriers: a task-anchored {@code INPUT_REQUIRED} status
* ({@code remoteTask} set) and a taskless bare-message Propose ({@code remoteTask == null},
- * pre-task A2A-T negotiation). The follow-up send reuses the remote task id when there is one
- * and stays taskless otherwise.
+ * pre-task A2A-T negotiation). The follow-up send reuses the remote task id when there is one and
+ * stays taskless otherwise.
*/
private CompletableFuture negotiate(
AgentCard card, Invocation invocation, SendMessageResult result, String remoteTask) {
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 d6d07f4..e54740d 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
@@ -181,16 +181,18 @@ 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");
+ 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));
}
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
index b820f26..b359519 100644
--- 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
@@ -120,7 +120,9 @@ public static RemoteA2AErrorException fromResponse(
/** 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()) {
+ 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 =
@@ -196,10 +198,7 @@ private static String safeText(JsonNode node, String key) {
/** 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("^_+|_+$", "");
+ reason.toLowerCase(Locale.ROOT).replaceAll("[^a-z0-9]+", "_").replaceAll("^_+|_+$", "");
if (!normalizedReason.isBlank()) return "a2a." + normalizedReason;
return "a2a.http." + httpStatus;
}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java
index 1b50af2..508a42d 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java
@@ -23,7 +23,9 @@
import java.util.Locale;
import java.util.concurrent.CancellationException;
-/** Classifies transport termination without making provider-specific messages the primary signal. */
+/**
+ * Classifies transport termination without making provider-specific messages the primary signal.
+ */
final class TransportFailures {
private TransportFailures() {}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/package-info.java
index 0f0ff69..20bb65d 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/package-info.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/package-info.java
@@ -18,6 +18,7 @@
*/
/**
- * Protocol adaptation: transport runtimes, authentication, negotiation correlation, independent subscriptions, and wire observability. Content generation stays with the host.
+ * Protocol adaptation: transport runtimes, authentication, negotiation correlation, independent
+ * subscriptions, and wire observability. Content generation stays with the host.
*/
package dev.openan.workflow.engine.client;
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/package-info.java
index 7c8c867..f0c2246 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/package-info.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/package-info.java
@@ -18,6 +18,7 @@
*/
/**
- * The four business callback contracts (onTask, onSelfTask, onRoute, onNegotiation) plus the engine event callback surface.
+ * The four business callback contracts (onTask, onSelfTask, onRoute, onNegotiation) plus the engine
+ * event callback surface.
*/
package dev.openan.workflow.engine.control;
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/package-info.java
index b4d4bfb..d9ac0b0 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/package-info.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/package-info.java
@@ -18,6 +18,7 @@
*/
/**
- * The scheduler: workflow graph validation, traversal, parallel dispatch, upstream context windows, and failure mapping.
+ * The scheduler: workflow graph validation, traversal, parallel dispatch, upstream context windows,
+ * and failure mapping.
*/
package dev.openan.workflow.engine.core;
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/BusinessInput.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/BusinessInput.java
index 1750964..f39bbeb 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/BusinessInput.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/BusinessInput.java
@@ -19,7 +19,6 @@
package dev.openan.workflow.engine.model;
-
/** Business content, not a protocol message. Exactly one of text/data is present. */
public record BusinessInput(String text, Object data) {
public BusinessInput {
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/ExecutionResult.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/ExecutionResult.java
index 0bf5a27..645bb43 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/ExecutionResult.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/ExecutionResult.java
@@ -36,10 +36,13 @@
@Builder
public class ExecutionResult {
private boolean success;
+
/** Per-step records in execution order. */
private List> history;
+
/** Step name to business outputs; nested structure preserved. */
private Map> stepOutputs;
+
/** Terminal failure description; raw error text, not redacted. Null on success. */
private String error;
}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/NegotiationRequest.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/NegotiationRequest.java
index bab21e9..e628983 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/NegotiationRequest.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/NegotiationRequest.java
@@ -25,8 +25,8 @@
/**
* Uninterpreted received content and history for one task's current negotiation session. Throws
- * {@link NullPointerException} if any component is null (including {@code previousExchanges})
- * and {@link IllegalArgumentException} if {@code remainingWait} is negative.
+ * {@link NullPointerException} if any component is null (including {@code previousExchanges}) and
+ * {@link IllegalArgumentException} if {@code remainingWait} is negative.
*/
public record NegotiationRequest(
TaskRequest task,
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java
index 22fee55..5e30a42 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java
@@ -24,37 +24,27 @@
* activated together with every unconditional edge and every other allowed conditional edge.
*/
public record RouteDecision(boolean allowed, String reason) {
- /**
- * Creates a decision and normalizes a null reason to an empty string.
- */
+ /** Creates a decision and normalizes a null reason to an empty string. */
public RouteDecision {
reason = reason == null ? "" : reason;
}
- /**
- * Allows the edge without an explanatory reason.
- */
+ /** Allows the edge without an explanatory reason. */
public static RouteDecision allow() {
return new RouteDecision(true, "");
}
- /**
- * Allows the edge and records the business reason for observability.
- */
+ /** Allows the edge and records the business reason for observability. */
public static RouteDecision allow(String reason) {
return new RouteDecision(true, reason);
}
- /**
- * Denies the edge without an explanatory reason.
- */
+ /** Denies the edge without an explanatory reason. */
public static RouteDecision deny() {
return new RouteDecision(false, "");
}
- /**
- * Denies the edge and records the business reason for observability.
- */
+ /** Denies the edge and records the business reason for observability. */
public static RouteDecision deny(String reason) {
return new RouteDecision(false, reason);
}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java
index 400929f..5028fb8 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java
@@ -22,9 +22,7 @@
import java.util.List;
import java.util.Objects;
-/**
- * Business input for evaluating one conditional outgoing edge.
- */
+/** Business input for evaluating one conditional outgoing edge. */
public record RouteRequest(
String executionId,
String stepName,
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/SendMessageResult.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/SendMessageResult.java
index 702267e..e789b64 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/SendMessageResult.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/SendMessageResult.java
@@ -32,8 +32,8 @@
* parts, message/task metadata, and artifacts at their own levels — and is the source for all
* convenience projections. It is not a wire-level transcript: envelope fields such as task and
* context identifiers, status history, and artifact delta flags are not preserved. {@code
- * failureCode}/{@code failureMessage} describe local interaction failures, separate from the
- * remote task state.
+ * failureCode}/{@code failureMessage} describe local interaction failures, separate from the remote
+ * task state.
*/
@Data
@NoArgsConstructor
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Task.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Task.java
index 444f3b7..7b6ea09 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Task.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Task.java
@@ -26,8 +26,8 @@
/**
* One dispatchable unit inside a {@link WorkflowStep}: the target agent and its business input.
- * Definitions are shared with the host, but each executor works on its own snapshot, so the
- * {@code status} field on the caller's instance is never mutated by a run.
+ * Definitions are shared with the host, but each executor works on its own snapshot, so the {@code
+ * status} field on the caller's instance is never mutated by a run.
*/
@Data
@NoArgsConstructor
@@ -36,12 +36,16 @@
public class Task {
/** Target agent name, resolved against the configured agent cards. */
private String agent;
+
/** Business input forwarded to callbacks through {@link TaskRequest}. */
private BusinessInput input;
+
/** Skill identifier used for events and logs. */
@Builder.Default private String skill = "";
+
/** Human-readable description used for events and logs. */
@Builder.Default private String description = "";
+
/** Runtime status; tracked on the executor's private snapshot, not on the caller's copy. */
@Builder.Default private TaskStatus status = TaskStatus.PENDING;
}
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/TaskResult.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/TaskResult.java
index d51aa5a..dfde9bc 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/TaskResult.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/TaskResult.java
@@ -27,9 +27,9 @@
/**
* Result of one dispatched task as seen by the workflow layer: success flag, ordered business
* outputs, and failure diagnostics. Remote evidence is retained in {@code receivedMessages} as
- * structured snapshots (parts, metadata, artifacts); envelope fields such as identifiers and
- * status history are not preserved. When evidence is present, the convenience {@code outputs}
- * are derived from it.
+ * structured snapshots (parts, metadata, artifacts); envelope fields such as identifiers and status
+ * history are not preserved. When evidence is present, the convenience {@code outputs} are derived
+ * from it.
*/
@Value
@Builder
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java
index d522ee3..6b6d8f0 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java
@@ -19,13 +19,12 @@
package dev.openan.workflow.engine.model;
+import java.util.List;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
-import java.util.List;
-
/**
* One node of a {@link Workflow}: the subtasks to dispatch, the outgoing jumps, and the upstream
* window its callbacks receive.
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/package-info.java
index eb13e4e..a825a35 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/package-info.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/package-info.java
@@ -18,6 +18,7 @@
*/
/**
- * Business-facing contracts: workflow definitions, callback request and response types, and message content. Depends only on standard A2A value types, never on A2A-T content-generation types.
+ * Business-facing contracts: workflow definitions, callback request and response types, and message
+ * content. Depends only on standard A2A value types, never on A2A-T content-generation types.
*/
package dev.openan.workflow.engine.model;
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java
index d53d0d2..0180d9b 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/LoadPsop.java
@@ -45,9 +45,10 @@
* {@code search} -- POST /api/v1/orchestrate/search (summary list by intent)
*
*
- * Defaults verify the server using JVM trust and hostname checks. Explicit {@code sslVerify=false}
- * skips both checks for this connection only, for controlled development without local trust files.
- * It does not remove the server's HTTPS certificate or satisfy a server requirement for mTLS.
+ *
Defaults verify the server using JVM trust and hostname checks. Explicit {@code
+ * sslVerify=false} skips both checks for this connection only, for controlled development without
+ * local trust files. It does not remove the server's HTTPS certificate or satisfy a server
+ * requirement for mTLS.
*/
public class LoadPsop {
private static final Logger log = LoggerFactory.getLogger(LoadPsop.class);
@@ -72,7 +73,10 @@ public static Workflow load(
.append(URLEncoder.encode(accessToken, StandardCharsets.UTF_8));
}
String url = urlBuilder.toString();
- log.info("[Registry] Loading PSOP from {} (ssl_verify={})", anonymousUrl(url, accessToken), sslVerify);
+ log.info(
+ "[Registry] Loading PSOP from {} (ssl_verify={})",
+ anonymousUrl(url, accessToken),
+ sslVerify);
HttpResult resp = execute("GET", url, null, sslVerify, timeouts);
if (resp.statusCode() != 200) {
throw requestFailure(resp);
@@ -112,7 +116,11 @@ public static List search(
.append(URLEncoder.encode(accessToken, StandardCharsets.UTF_8));
}
String url = urlBuilder.toString();
- log.info("[Registry] Searching PSOP at {} (intent={}, top_n={})", anonymousUrl(url, accessToken), intent, topN);
+ log.info(
+ "[Registry] Searching PSOP at {} (intent={}, top_n={})",
+ anonymousUrl(url, accessToken),
+ intent,
+ topN);
String jsonBody = mapper.writeValueAsString(Map.of("intent", intent, "top_n", topN));
HttpResult resp = execute("POST", url, jsonBody, sslVerify, timeouts);
if (resp.statusCode() != 200) {
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/RegistryClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/RegistryClient.java
index 002546b..ae24596 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/RegistryClient.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/registry/RegistryClient.java
@@ -81,7 +81,8 @@ public Map fetchAgentCard(String name) throws Exception {
public List> fetchAgentCards() throws Exception {
String url = baseUrl + "/rest/v1/registry-center/agent-cards";
log.info("[Registry] Fetching all agent cards from {}", baseUrl);
- HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build();
+ HttpRequest request =
+ HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build();
HttpResponse resp = send(request);
if (resp.statusCode() != 200) {
throw requestFailure("Registry", resp.statusCode(), resp.body());
@@ -116,7 +117,8 @@ public Map fetchAgentCard(String name, String organization) thro
}
String url = urlBuilder.toString();
log.info("[Registry] Fetching agent card: name={}, org={}", name, organization);
- HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build();
+ HttpRequest request =
+ HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build();
HttpResponse resp = send(request);
if (resp.statusCode() != 200) {
throw requestFailure("Registry", resp.statusCode(), resp.body());
@@ -165,7 +167,8 @@ private HttpResponse send(HttpRequest request) throws Exception {
var response = httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString());
try {
// Also bound body consumption: a peer may send headers and then stall its body.
- return response.get(Math.max(1, requestTimeout.toMillis()), java.util.concurrent.TimeUnit.MILLISECONDS);
+ return response.get(
+ Math.max(1, requestTimeout.toMillis()), java.util.concurrent.TimeUnit.MILLISECONDS);
} catch (java.util.concurrent.TimeoutException error) {
response.cancel(true);
throw new java.net.http.HttpTimeoutException("Registry response deadline exceeded");
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java
index 31a9581..5276be8 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java
@@ -128,32 +128,37 @@ public static CompletableFuture execute(
}
CompletableFuture exposed = new CompletableFuture<>();
var closed = new java.util.concurrent.atomic.AtomicBoolean();
- Runnable cleanup = () -> {
- if (closed.compareAndSet(false, true)) {
- closeExecution(client, closeClientOnFinish, collectingCallback);
- }
- };
- var finishing = new java.util.concurrent.atomic.AtomicReference>();
- exposed.whenComplete((result, error) -> {
- if (exposed.isCancelled()) {
- execution.cancel(true);
- CompletableFuture pending = finishing.get();
- if (pending != null) pending.cancel(true);
- cleanup.run();
- }
- });
+ Runnable cleanup =
+ () -> {
+ if (closed.compareAndSet(false, true)) {
+ closeExecution(client, closeClientOnFinish, collectingCallback);
+ }
+ };
+ var finishing =
+ new java.util.concurrent.atomic.AtomicReference>();
+ exposed.whenComplete(
+ (result, error) -> {
+ if (exposed.isCancelled()) {
+ execution.cancel(true);
+ CompletableFuture pending = finishing.get();
+ if (pending != null) pending.cancel(true);
+ cleanup.run();
+ }
+ });
// Observe the source independently: cancelling the public future must not skip cleanup.
- execution.whenComplete((result, error) -> {
- ExecutionResult outcome = error == null ? result : handleExecutionError(error);
- CompletableFuture completion =
- finalizeResult(outcome, collectingCallback, collected, onFinish, cleanup);
- finishing.set(completion);
- if (exposed.isCancelled()) completion.cancel(true);
- completion.whenComplete((value, failure) -> {
- if (failure == null) exposed.complete(value);
- else exposed.completeExceptionally(failure);
- });
- });
+ execution.whenComplete(
+ (result, error) -> {
+ ExecutionResult outcome = error == null ? result : handleExecutionError(error);
+ CompletableFuture completion =
+ finalizeResult(outcome, collectingCallback, collected, onFinish, cleanup);
+ finishing.set(completion);
+ if (exposed.isCancelled()) completion.cancel(true);
+ completion.whenComplete(
+ (value, failure) -> {
+ if (failure == null) exposed.complete(value);
+ else exposed.completeExceptionally(failure);
+ });
+ });
return exposed;
}
@@ -282,8 +287,10 @@ private static WorkflowEngineClient createEngineClient(
private static ExecutionResult handleExecutionError(Throwable error) {
boolean cancelled = error instanceof java.util.concurrent.CancellationException;
- String message = cancelled ? "Workflow execution cancelled"
- : error.getMessage() != null ? error.getMessage() : error.getClass().getSimpleName();
+ String message =
+ cancelled
+ ? "Workflow execution cancelled"
+ : error.getMessage() != null ? error.getMessage() : error.getClass().getSimpleName();
if (cancelled) log.info("[execute_psop] {}", message);
else log.error("[execute_psop] Execution failed: {}", message);
return ExecutionResult.builder()
diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/util/package-info.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/util/package-info.java
index 9a09850..04ec494 100644
--- a/workflow-engine/src/main/java/dev/openan/workflow/engine/util/package-info.java
+++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/util/package-info.java
@@ -17,7 +17,5 @@
* under the License.
*/
-/**
- * Cross-cutting helpers shared by engine packages, such as sensitive-data redaction.
- */
+/** Cross-cutting helpers shared by engine packages, such as sensitive-data redaction. */
package dev.openan.workflow.engine.util;
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
index c4ae363..2cd7044 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2AErrorDetectingHttpClientTest.java
@@ -38,20 +38,27 @@ void standardErrorCancelsUpstreamAndSurvivesAThrowingErrorObserver() throws Exce
var failures = new AtomicInteger();
var events = new AtomicInteger();
var completions = new AtomicInteger();
- var client = client((messages, errors, complete) -> {
- messages.accept(new ServerSentEvent(RemoteA2AErrorResponseTest.error(429)));
- messages.accept(new ServerSentEvent("{\"message\":{}}"));
- complete.run();
- errors.accept(new IllegalStateException("late close"));
- return upstream;
- });
- var result = client.createPost().postAsyncSSE(event -> events.incrementAndGet(),
- error -> {
- failures.incrementAndGet();
- throw new IllegalStateException("observer failed");
- }, completions::incrementAndGet);
- var remoteError = RemoteA2AErrorException.findIn(
- assertThrows(CompletionException.class, result::join));
+ var client =
+ client(
+ (messages, errors, complete) -> {
+ messages.accept(new ServerSentEvent(RemoteA2AErrorResponseTest.error(429)));
+ messages.accept(new ServerSentEvent("{\"message\":{}}"));
+ complete.run();
+ errors.accept(new IllegalStateException("late close"));
+ return upstream;
+ });
+ var result =
+ client
+ .createPost()
+ .postAsyncSSE(
+ event -> events.incrementAndGet(),
+ error -> {
+ failures.incrementAndGet();
+ throw new IllegalStateException("observer failed");
+ },
+ completions::incrementAndGet);
+ var remoteError =
+ RemoteA2AErrorException.findIn(assertThrows(CompletionException.class, result::join));
assertNotNull(remoteError);
assertEquals(429, remoteError.getHttpStatus());
assertEquals(1, remoteError.getSuppressed().length);
@@ -67,12 +74,18 @@ void cancellationPropagatesAndLateEventsCannotCompleteTheCall() throws Exception
var messages = new AtomicReference>();
var complete = new AtomicReference();
var callbacks = new AtomicInteger();
- var result = client((m, e, c) -> {
- messages.set(m);
- complete.set(c);
- return upstream;
- }).createPost().postAsyncSSE(event -> callbacks.incrementAndGet(),
- error -> callbacks.incrementAndGet(), callbacks::incrementAndGet);
+ var result =
+ client(
+ (m, e, c) -> {
+ messages.set(m);
+ complete.set(c);
+ return upstream;
+ })
+ .createPost()
+ .postAsyncSSE(
+ event -> callbacks.incrementAndGet(),
+ error -> callbacks.incrementAndGet(),
+ callbacks::incrementAndGet);
assertTrue(result.cancel(true));
assertTrue(upstream.isCancelled());
messages.get().accept(new ServerSentEvent("{}"));
@@ -82,16 +95,23 @@ void cancellationPropagatesAndLateEventsCannotCompleteTheCall() throws Exception
@Test
void successfulDataIsNotRewrittenOrMistakenForANestedError() throws Exception {
- String data = "{\"task\":{\"status\":{\"state\":\"TASK_STATE_COMPLETED\"}},"
- + "\"metadata\":{\"error\":{\"code\":400,\"message\":\"business value\"}}}";
+ 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) -> {
- messages.accept(new ServerSentEvent(data));
- complete.run();
- return CompletableFuture.completedFuture(null);
- }).createPost().postAsyncSSE(event -> seen.set(event.data()),
- error -> fail("unexpected error", error), completions::incrementAndGet);
+ var result =
+ client(
+ (messages, errors, complete) -> {
+ messages.accept(new ServerSentEvent(data));
+ complete.run();
+ return CompletableFuture.completedFuture(null);
+ })
+ .createPost()
+ .postAsyncSSE(
+ event -> seen.set(event.data()),
+ error -> fail("unexpected error", error),
+ completions::incrementAndGet);
result.join();
assertEquals(data, seen.get());
assertEquals(1, completions.get());
@@ -101,11 +121,15 @@ void successfulDataIsNotRewrittenOrMistakenForANestedError() throws Exception {
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 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());
@@ -115,28 +139,61 @@ void streamThatCompletesWithoutAnA2AEventFailsImmediately() throws Exception {
}
private static A2AHttpClient client(Start start) {
- return new A2AErrorDetectingHttpClient(new A2AHttpClient() {
- @Override public GetBuilder createGet() { throw new UnsupportedOperationException(); }
- @Override public DeleteBuilder createDelete() { throw new UnsupportedOperationException(); }
- @Override public PostBuilder createPost() {
- return new PostBuilder() {
- @Override public PostBuilder url(String url) { return this; }
- @Override public PostBuilder body(String body) { return this; }
- @Override public PostBuilder addHeader(String key, String value) { return this; }
- @Override public PostBuilder addHeaders(Map values) { return this; }
- @Override public A2AHttpResponse post() { throw new UnsupportedOperationException(); }
- @Override public CompletableFuture postAsyncSSE(
- Consumer messages, Consumer errors, Runnable complete) {
- return start.run(messages, errors, complete);
+ return new A2AErrorDetectingHttpClient(
+ new A2AHttpClient() {
+ @Override
+ public GetBuilder createGet() {
+ throw new UnsupportedOperationException();
}
- };
- }
- });
+
+ @Override
+ public DeleteBuilder createDelete() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public PostBuilder createPost() {
+ return new PostBuilder() {
+ @Override
+ public PostBuilder url(String url) {
+ return this;
+ }
+
+ @Override
+ public PostBuilder body(String body) {
+ return this;
+ }
+
+ @Override
+ public PostBuilder addHeader(String key, String value) {
+ return this;
+ }
+
+ @Override
+ public PostBuilder addHeaders(Map values) {
+ return this;
+ }
+
+ @Override
+ public A2AHttpResponse post() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public CompletableFuture postAsyncSSE(
+ Consumer messages,
+ Consumer errors,
+ Runnable complete) {
+ return start.run(messages, errors, complete);
+ }
+ };
+ }
+ });
}
@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/A2ATransportHeaderTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java
index c4c549d..46b20cb 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/A2ATransportHeaderTest.java
@@ -161,14 +161,11 @@ void forwardsBusinessSelectedNegotiationActivationWithoutNegotiationMetadata() t
extensions);
try (A2ATransport transport =
- new A2ATransport(
- List.of(card), runtime, WorkflowEngineClientConfig.builder().build())) {
+ new A2ATransport(List.of(card), runtime, WorkflowEngineClientConfig.builder().build())) {
transport.send(card, card.name(), message, "context-1", null, null).join();
}
- assertEquals(
- TASK_T_URI + "," + NEGOTIATION_T_URI,
- capturedHeaders.get().get("A2A-Extensions"));
+ assertEquals(TASK_T_URI + "," + NEGOTIATION_T_URI, capturedHeaders.get().get("A2A-Extensions"));
}
@Test
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java
index 330fdc4..0136eb5 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java
@@ -55,10 +55,14 @@ private static AgentCard card(String requirements) throws Exception {
private static Map scheme(String url, String header) {
return Map.of(
- "login_url", url,
- "token_field", "access_token",
- "request_fields", Map.of("username", "user", "password", "password"),
- "auth_header", header);
+ "login_url",
+ url,
+ "token_field",
+ "access_token",
+ "request_fields",
+ Map.of("username", "user", "password", "password"),
+ "auth_header",
+ header);
}
@Test
@@ -102,10 +106,7 @@ void satisfiesEverySchemeInAnAndRequirement() throws Exception {
Map headers =
contributor.contribute(
- card("[{\"schemes\":{\"first\":{},\"second\":{}}}]"),
- "agent",
- Map.of(),
- Map.of());
+ card("[{\"schemes\":{\"first\":{},\"second\":{}}}]"), "agent", Map.of(), Map.of());
assertEquals(Map.of("X-First", "token", "X-Second", "token"), headers);
} finally {
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java
index 2529735..fb1411b 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/DefaultWorkflowEngineClientNegotiationTest.java
@@ -300,7 +300,7 @@ public void close() {}
CompletionException error =
assertThrows(
CompletionException.class,
- () -> client.sendTask("test", negotiationTask("start")).join());
+ () -> client.sendTask("test", negotiationTask("start")).join());
assertInstanceOf(TimeoutException.class, error.getCause());
assertTrue(canceled.await(3, TimeUnit.SECONDS));
}
@@ -343,7 +343,7 @@ public void close() {}
CompletionException error =
assertThrows(
CompletionException.class,
- () -> client.sendTask("test", negotiationTask("start")).join());
+ () -> client.sendTask("test", negotiationTask("start")).join());
assertTrue(error.getCause().getMessage().contains("TASK_STATE_AUTH_REQUIRED"));
assertEquals(1, cancellations.get());
}
@@ -443,7 +443,7 @@ public void close() {}
CompletionException error =
assertThrows(
CompletionException.class,
- () -> client.sendTask("test", negotiationTask("start")).join());
+ () -> client.sendTask("test", negotiationTask("start")).join());
RemoteA2AErrorException actual = RemoteA2AErrorException.findIn(error);
assertNotNull(actual);
assertEquals(429, actual.getHttpStatus());
@@ -485,7 +485,7 @@ void answersFinalRoundAndKeepsOriginalRemoteAssociation() throws Exception {
.build());
assertEquals(
"TASK_STATE_COMPLETED",
- client.sendTask("test", negotiationTask("start")).join().getTaskState());
+ client.sendTask("test", negotiationTask("start")).join().getTaskState());
}
assertEquals(2, sent.size());
assertNull(sent.get(0).message().taskId());
@@ -519,7 +519,7 @@ void terminalStateNeverRestartsOldNegotiation() throws Exception {
return CompletableFuture.failedFuture(new AssertionError());
})
.build());
- client.sendTask("test", negotiationTask("start")).join();
+ client.sendTask("test", negotiationTask("start")).join();
}
assertEquals(0, callbacks.get());
}
@@ -552,7 +552,7 @@ void rejectsWrongContextNewProposeAndMissingCallbackWithoutSendingAgain() throws
.build());
assertThrows(
CompletionException.class,
- () -> client.sendTask("test", negotiationTask("start")).join());
+ () -> client.sendTask("test", negotiationTask("start")).join());
}
assertEquals(1, sends.get());
}
@@ -629,13 +629,16 @@ void tasklessRoundLoopAdvancesRoundsAndCompletesWithTaskAnchoredResult() throws
params -> {
sent.add(params);
return switch (sent.size()) {
- case 1 -> messageResponse(
- params, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE));
- case 2 -> messageResponse(
- params, negotiation("pre", 2, 5, NegotiationPerformative.PROPOSE));
- default -> List.of(
- response(
- params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done")));
+ case 1 ->
+ messageResponse(
+ params, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE));
+ case 2 ->
+ messageResponse(
+ params, negotiation("pre", 2, 5, NegotiationPerformative.PROPOSE));
+ default ->
+ List.of(
+ response(
+ params, TaskState.TASK_STATE_COMPLETED, MessageContent.text("done")));
};
});
try (var transport =
@@ -716,9 +719,7 @@ void duplicateTasklessProposeFailsExplicitlyWithoutRepeatingTheCallback() throws
new A2ATransport(
List.of(card()),
runtime(
- p ->
- messageResponse(
- p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE))),
+ p -> messageResponse(p, negotiation("pre", 1, 5, NegotiationPerformative.PROPOSE))),
WorkflowEngineClientConfig.builder().build())) {
var client = new DefaultWorkflowEngineClient(transport);
client.setControlPoint(
@@ -794,7 +795,7 @@ void abortAcknowledgementIsNotDiagnosisSuccessAndPreservesRemoteState() throws E
new NegotiationReply.Send(
negotiation("city1", 1, NegotiationPerformative.ABORT))))
.build());
- var result = client.sendTask("test", negotiationTask("start")).join();
+ var result = client.sendTask("test", negotiationTask("start")).join();
assertEquals("TASK_STATE_COMPLETED", result.getTaskState());
assertEquals("negotiation.aborted", result.getFailureCode());
}
@@ -857,7 +858,7 @@ public void close() {}
assertEquals(
"TASK_STATE_COMPLETED",
client
- .sendTask("test", negotiationTask("start"))
+ .sendTask("test", negotiationTask("start"))
.get(5, TimeUnit.SECONDS)
.getTaskState());
assertEquals(2, sends.get());
@@ -895,7 +896,7 @@ void canceledOrTimedOutNegotiationNeverSendsLateReply() throws Exception {
return answer;
})
.build());
- var result = client.sendTask("test", negotiationTask("start"));
+ var result = client.sendTask("test", negotiationTask("start"));
assertTrue(entered.await(1, TimeUnit.SECONDS));
if (timeout) assertThrows(ExecutionException.class, () -> result.get(3, TimeUnit.SECONDS));
else assertTrue(result.cancel(true));
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 9e734f0..c700c00 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
@@ -313,15 +313,50 @@ public CompletableFuture> sendAsync(
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(); }
+ @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 {
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
index 04be1a9..7e6c404 100644
--- 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
@@ -42,19 +42,25 @@ 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
+ 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 + "\","
+ + reason
+ + "\",\"domain\":\""
+ + domain
+ + "\","
+ "\"metadata\":{\"taskId\":\"task-123\"}}]}}";
}
@Test
void preservesStandardFieldsAndFindsErrorInfoAmongTypedDetails() {
- var error = RemoteA2AErrorException.fromResponse(
- 400, error(400), Map.of("Retry-After", List.of("5")));
+ var error =
+ RemoteA2AErrorException.fromResponse(400, error(400), Map.of("Retry-After", List.of("5")));
assertNotNull(error);
assertEquals(400, error.getHttpStatus());
assertEquals(400, error.getCode());
@@ -69,29 +75,33 @@ void preservesStandardFieldsAndFindsErrorInfoAmongTypedDetails() {
@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(
+ "{\"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"));
+ 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");
+ String payload = error(400).replace("INVALID_PARAMS", "Invalid Params / Port#1");
var error = RemoteA2AErrorException.fromPayload(payload);
@@ -106,50 +116,74 @@ void directRuntimeFailsPromptlyWhenSdkExposesAnErrorBodyAsSseData(int status) th
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.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();
+ 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 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");
+ assertTrue(
+ disconnected.await(3, TimeUnit.SECONDS), "failed SSE must close its HTTP connection");
} finally {
release.countDown();
runtime.close();
@@ -159,8 +193,9 @@ void directRuntimeFailsPromptlyWhenSdkExposesAnErrorBodyAsSseData(int status) th
@Test
void projectsTypedSdkErrorsAndHandlesCauseCycles() {
- var projected = RemoteA2AErrorException.findIn(
- new IllegalStateException("wrapper", new InvalidParamsError("missing port")));
+ var projected =
+ RemoteA2AErrorException.findIn(
+ new IllegalStateException("wrapper", new InvalidParamsError("missing port")));
assertNotNull(projected);
assertEquals("INVALID_PARAMS", projected.getReason());
assertEquals("a2a.invalid_params", projected.workflowErrorCode());
@@ -174,18 +209,28 @@ void projectsTypedSdkErrorsAndHandlesCauseCycles() {
@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.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());
+ 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 {
@@ -196,27 +241,48 @@ void nonStreamingResponsesPreserveTheStandardA2AError() throws Exception {
@Test
void runtimeUsesTheHttpClientCustomizationHookWithoutBypassingErrorDetection() throws Exception {
HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
- server.createContext("/", exchange -> {
- exchange.getRequestBody().readAllBytes();
- byte[] body = ("data: " + error(400) + "\n\n").getBytes(StandardCharsets.UTF_8);
- exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
- exchange.sendResponseHeaders(200, 0);
- exchange.getResponseBody().write(body);
- exchange.close();
- });
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ byte[] body = ("data: " + error(400) + "\n\n").getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "text/event-stream");
+ exchange.sendResponseHeaders(200, 0);
+ exchange.getResponseBody().write(body);
+ exchange.close();
+ });
server.start();
var runtime = new CustomizingRuntime();
try {
- AgentCard card = AgentCard.builder().name("customized-runtime").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 thrown = assertThrows(RuntimeException.class,
- () -> runtime.sendMessage(card, params, null, ignored -> {}, null));
+ AgentCard card =
+ AgentCard.builder()
+ .name("customized-runtime")
+ .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 thrown =
+ assertThrows(
+ RuntimeException.class,
+ () -> runtime.sendMessage(card, params, null, ignored -> {}, null));
assertNotNull(RemoteA2AErrorException.findIn(thrown));
assertEquals(1, runtime.customizations.get());
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java
index d50757c..11d4542 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java
@@ -33,6 +33,7 @@ void prefersExceptionTypeAndStillSupportsKnownProviderFallbacks() {
new RuntimeException("wrapper", new AsynchronousCloseException())));
assertTrue(
TransportFailures.isExpectedLocalClose(new IOException("Connection closed locally")));
- assertFalse(TransportFailures.isExpectedLocalClose(new IOException("connection reset by peer")));
+ assertFalse(
+ TransportFailures.isExpectedLocalClose(new IOException("connection reset by peer")));
}
}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java
index 47b23a4..efaeb4e 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java
@@ -18,16 +18,15 @@
*/
package dev.openan.workflow.engine.control;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
import dev.openan.workflow.engine.model.RouteRequest;
import dev.openan.workflow.engine.model.TaskRequest;
import dev.openan.workflow.engine.model.TaskResult;
import dev.openan.workflow.engine.model.WorkflowInput;
-import org.junit.jupiter.api.Test;
-
import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
+import org.junit.jupiter.api.Test;
class DefaultControlPointTest {
@@ -55,8 +54,8 @@ void conditionalRoutingRequiresBusinessHandler() {
() ->
new DefaultControlPoint()
.onRoute(
- new RouteRequest(
- "run", "step", "next", "condition", WorkflowInput.empty(), List.of()))
+ new RouteRequest(
+ "run", "step", "next", "condition", WorkflowInput.empty(), List.of()))
.join());
}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
index e9eb32a..d90faf7 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/RemoteA2AErrorFailureTest.java
@@ -37,10 +37,14 @@ 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
+ 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\"}]}}";
+ + reason
+ + "\",\"domain\":\"example.invalid\"}]}}";
}
@ParameterizedTest
@@ -64,38 +68,60 @@ public CompletableFuture sendTask(
}
};
var merges = new AtomicInteger();
- var callbacks = ControlPoint.builder()
- .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(
- 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();
- var events = new dev.openan.workflow.engine.control.EventCallback() {
- @Override public void onEvent(String type, Map data) {
- if (type.equals(dev.openan.workflow.engine.control.EventType.TASK_RESPONSE)) {
- observed.add(data);
- if (Boolean.FALSE.equals(data.get("success"))) reported.countDown();
- }
- if (type.equals(dev.openan.workflow.engine.control.EventType.ERROR)) {
- assertNotNull(data.get("error"));
- }
- }
- };
- var execution = new WorkflowExecutor(workflow, callbacks, client, events, "request", "en").run();
+ var callbacks =
+ ControlPoint.builder()
+ .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(
+ 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();
+ var events =
+ new dev.openan.workflow.engine.control.EventCallback() {
+ @Override
+ public void onEvent(String type, Map data) {
+ if (type.equals(dev.openan.workflow.engine.control.EventType.TASK_RESPONSE)) {
+ observed.add(data);
+ if (Boolean.FALSE.equals(data.get("success"))) reported.countDown();
+ }
+ if (type.equals(dev.openan.workflow.engine.control.EventType.ERROR)) {
+ assertNotNull(data.get("error"));
+ }
+ }
+ };
+ var execution =
+ new WorkflowExecutor(workflow, callbacks, client, events, "request", "en").run();
try {
assertTrue(started.await(5, java.util.concurrent.TimeUnit.SECONDS));
agent1.completeExceptionally(RemoteA2AErrorException.fromPayload(error(status)));
assertTrue(reported.await(5, java.util.concurrent.TimeUnit.SECONDS));
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("agent2 result"), Map.of(), List.of()))).build());
+ agent2.complete(
+ SendMessageResult.builder()
+ .taskState("TASK_STATE_COMPLETED")
+ .receivedMessages(
+ List.of(
+ new ReceivedMessage(
+ 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());
@@ -108,8 +134,11 @@ public CompletableFuture sendTask(
}
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("agent2")).findFirst().orElseThrow();
+ var success =
+ result.getHistory().stream()
+ .filter(h -> h.get("agent").equals("agent2"))
+ .findFirst()
+ .orElseThrow();
assertEquals("success", success.get("status"));
assertEquals(List.of("agent2 result"), success.get("outputs"));
} finally {
@@ -120,9 +149,12 @@ public CompletableFuture sendTask(
}
private static WorkflowStep remote(String name) {
- return WorkflowStep.builder().name(name).stepType(StepType.ALL_SUCCESS)
+ return WorkflowStep.builder()
+ .name(name)
+ .stepType(StepType.ALL_SUCCESS)
.subtasks(List.of(Task.builder().agent(name).description("execute").build()))
- .next(List.of(new JumpCondition("merge", ""))).build();
+ .next(List.of(new JumpCondition("merge", "")))
+ .build();
}
@ParameterizedTest
@@ -142,23 +174,32 @@ public CompletableFuture sendTask(
return CompletableFuture.failedFuture(new CompletionException(remoteError));
}
};
- 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 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("execute")))
- .onNegotiation(request -> {
- negotiations.incrementAndGet();
- return CompletableFuture.failedFuture(new AssertionError("must not negotiate"));
- }).build();
- var result = new WorkflowExecutor(workflow, callbacks, client, null, "request", "en").run().join();
+ var callbacks =
+ ControlPoint.builder()
+ .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, "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(status == 429 ? "a2a.active_task_limit_exceeded" : "a2a.invalid_params",
+ 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("rejected"));
@@ -168,12 +209,16 @@ public CompletableFuture sendTask(
@Test
void knownSecretsAreRedactedAndUnrecognizedProviderErrorsRemainPrivate() {
- var remoteError = RemoteA2AErrorException.fromPayload(error(400).replace(
- "Remote agent rejected the request", "bad request password=private-value"));
+ 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(remoteError.getMessage().contains("private-value"));
- assertEquals("IllegalStateException",
+ assertEquals(
+ "IllegalStateException",
FailureMapping.from(new IllegalStateException("raw private provider body")).getError());
}
}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopHttpsTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopHttpsTest.java
index e5cdab6..24d7fc1 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopHttpsTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopHttpsTest.java
@@ -55,11 +55,13 @@ void explicitOptOutLoadsAndSearchesWithoutTrustFilesAndWithoutGlobalSideEffects(
var defaults = HttpsURLConnection.getDefaultSSLSocketFactory();
var verifier = HttpsURLConnection.getDefaultHostnameVerifier();
var defaultContext = SSLContext.getDefault();
- String hostnameProperty = System.getProperty("jdk.internal.httpclient.disableHostnameVerification");
+ String hostnameProperty =
+ System.getProperty("jdk.internal.httpclient.disableHostnameVerification");
try (var endpoint = endpoint(wrongSan, false)) {
String baseUrl = endpoint.baseUrl();
assertEquals("tls-psop", LoadPsop.load(baseUrl, "tls-psop", null, false).getName());
- assertEquals("tls-psop", LoadPsop.search(baseUrl, "complaint", 3, null, false).get(0).getWorkflowId());
+ assertEquals(
+ "tls-psop", LoadPsop.search(baseUrl, "complaint", 3, null, false).get(0).getWorkflowId());
assertThrows(SSLHandshakeException.class, () -> LoadPsop.load(baseUrl, "tls-psop"));
assertThrows(SSLHandshakeException.class, () -> LoadPsop.search(baseUrl, "complaint"));
@@ -77,30 +79,60 @@ void explicitOptOutLoadsAndSearchesWithoutTrustFilesAndWithoutGlobalSideEffects(
assertSame(defaults, HttpsURLConnection.getDefaultSSLSocketFactory());
assertSame(verifier, HttpsURLConnection.getDefaultHostnameVerifier());
assertSame(defaultContext, SSLContext.getDefault());
- assertEquals(hostnameProperty, System.getProperty("jdk.internal.httpclient.disableHostnameVerification"));
+ assertEquals(
+ hostnameProperty,
+ System.getProperty("jdk.internal.httpclient.disableHostnameVerification"));
}
@Test
void skippingServerVerificationDoesNotBypassMutualTls() throws Exception {
try (var endpoint = endpoint(false, true)) {
- assertThrows(IOException.class, () -> LoadPsop.load(endpoint.baseUrl(), "tls-psop", null, false));
+ assertThrows(
+ IOException.class, () -> LoadPsop.load(endpoint.baseUrl(), "tls-psop", null, false));
}
}
private Endpoint endpoint(boolean wrongSan, boolean clientAuth) throws Exception {
Path keyStoreFile = directory.resolve("server.p12");
- Path keytool = Path.of(System.getProperty("java.home"), "bin",
- System.getProperty("os.name").startsWith("Windows") ? "keytool.exe" : "keytool");
- var command = new ArrayList<>(List.of(keytool.toString(), "-genkeypair", "-alias", "test",
- "-keyalg", "RSA", "-keysize", "2048", "-dname", "CN=unrelated.example.test",
- "-validity", "2", "-storetype", "PKCS12", "-keystore", keyStoreFile.toString(),
- "-storepass", "test-only-password", "-keypass", "test-only-password", "-noprompt"));
+ Path keytool =
+ Path.of(
+ System.getProperty("java.home"),
+ "bin",
+ System.getProperty("os.name").startsWith("Windows") ? "keytool.exe" : "keytool");
+ var command =
+ new ArrayList<>(
+ List.of(
+ keytool.toString(),
+ "-genkeypair",
+ "-alias",
+ "test",
+ "-keyalg",
+ "RSA",
+ "-keysize",
+ "2048",
+ "-dname",
+ "CN=unrelated.example.test",
+ "-validity",
+ "2",
+ "-storetype",
+ "PKCS12",
+ "-keystore",
+ keyStoreFile.toString(),
+ "-storepass",
+ "test-only-password",
+ "-keypass",
+ "test-only-password",
+ "-noprompt"));
if (wrongSan) command.addAll(List.of("-ext", "SAN=dns:unrelated.example.test"));
- var process = new ProcessBuilder(command).redirectErrorStream(true)
- .redirectOutput(directory.resolve("keytool.log").toFile()).start();
+ var process =
+ new ProcessBuilder(command)
+ .redirectErrorStream(true)
+ .redirectOutput(directory.resolve("keytool.log").toFile())
+ .start();
try {
assertTrue(process.waitFor(20, TimeUnit.SECONDS), "keytool timed out");
- assertEquals(0, process.exitValue(), () -> "keytool failed; see " + directory.resolve("keytool.log"));
+ assertEquals(
+ 0, process.exitValue(), () -> "keytool failed; see " + directory.resolve("keytool.log"));
} finally {
if (process.isAlive()) process.destroyForcibly();
}
@@ -116,30 +148,44 @@ private Endpoint endpoint(boolean wrongSan, boolean clientAuth) throws Exception
var context = SSLContext.getInstance("TLS");
context.init(keys.getKeyManagers(), null, null);
var server = HttpsServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
- server.setHttpsConfigurator(new HttpsConfigurator(context) {
- @Override public void configure(HttpsParameters parameters) {
- var ssl = context.getDefaultSSLParameters();
- ssl.setNeedClientAuth(clientAuth);
- parameters.setSSLParameters(ssl);
- }
- });
- server.createContext("/", exchange -> {
- exchange.getRequestBody().readAllBytes();
- String body = exchange.getRequestURI().getPath().endsWith("/search")
- ? "{\"data\":[{\"workflow_id\":\"tls-psop\",\"name\":\"tls-psop\"}]}"
- : "{\"data\":{\"name\":\"tls-psop\",\"steps\":[]}}";
- byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
- exchange.getResponseHeaders().set("Content-Type", "application/json");
- exchange.sendResponseHeaders(200, bytes.length);
- try (var output = exchange.getResponseBody()) { output.write(bytes); }
- finally { exchange.close(); }
- });
+ server.setHttpsConfigurator(
+ new HttpsConfigurator(context) {
+ @Override
+ public void configure(HttpsParameters parameters) {
+ var ssl = context.getDefaultSSLParameters();
+ ssl.setNeedClientAuth(clientAuth);
+ parameters.setSSLParameters(ssl);
+ }
+ });
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.getRequestBody().readAllBytes();
+ String body =
+ exchange.getRequestURI().getPath().endsWith("/search")
+ ? "{\"data\":[{\"workflow_id\":\"tls-psop\",\"name\":\"tls-psop\"}]}"
+ : "{\"data\":{\"name\":\"tls-psop\",\"steps\":[]}}";
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(200, bytes.length);
+ try (var output = exchange.getResponseBody()) {
+ output.write(bytes);
+ } finally {
+ exchange.close();
+ }
+ });
server.start();
return new Endpoint(server);
}
private record Endpoint(HttpsServer server) implements AutoCloseable {
- String baseUrl() { return "https://127.0.0.1:" + server.getAddress().getPort(); }
- @Override public void close() { server.stop(0); }
+ String baseUrl() {
+ return "https://127.0.0.1:" + server.getAddress().getPort();
+ }
+
+ @Override
+ public void close() {
+ server.stop(0);
+ }
}
}
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java
index a8de6e4..a3ac1e6 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/LoadPsopSearchTest.java
@@ -83,9 +83,12 @@ void customReadTimeoutIsApplied() throws Exception {
@Test
void tokenLogsRevealPresenceButNoTokenCharacters() {
- assertEquals("https://example.test/search?access_token=",
- LoadPsop.anonymousUrl("https://example.test/search?access_token=secret%2Bvalue", "secret+value"));
- assertEquals("https://example.test/search", LoadPsop.anonymousUrl("https://example.test/search", null));
+ assertEquals(
+ "https://example.test/search?access_token=",
+ LoadPsop.anonymousUrl(
+ "https://example.test/search?access_token=secret%2Bvalue", "secret+value"));
+ assertEquals(
+ "https://example.test/search", LoadPsop.anonymousUrl("https://example.test/search", null));
}
/**
diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/RegistryClientTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/RegistryClientTest.java
index eb7e85f..5d7ed94 100644
--- a/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/RegistryClientTest.java
+++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/registry/RegistryClientTest.java
@@ -28,26 +28,37 @@
class RegistryClientTest {
@Test
void rejectsNonPositiveDeadline() {
- assertThrows(IllegalArgumentException.class, () -> new RegistryClient("http://localhost", true, Duration.ZERO));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> new RegistryClient("http://localhost", true, Duration.ZERO));
}
@Test
void deadlineIncludesStalledResponseBody() throws Exception {
var server = com.sun.net.httpserver.HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
var release = new java.util.concurrent.CountDownLatch(1);
- server.createContext("/", exchange -> {
- exchange.sendResponseHeaders(200, 100);
- exchange.getResponseBody().write('{');
- exchange.getResponseBody().flush();
- try { release.await(3, java.util.concurrent.TimeUnit.SECONDS); }
- catch (InterruptedException error) { Thread.currentThread().interrupt(); }
- finally { exchange.close(); }
- });
+ server.createContext(
+ "/",
+ exchange -> {
+ exchange.sendResponseHeaders(200, 100);
+ exchange.getResponseBody().write('{');
+ exchange.getResponseBody().flush();
+ try {
+ release.await(3, java.util.concurrent.TimeUnit.SECONDS);
+ } catch (InterruptedException error) {
+ Thread.currentThread().interrupt();
+ } finally {
+ exchange.close();
+ }
+ });
server.start();
try {
- var client = new RegistryClient("http://127.0.0.1:" + server.getAddress().getPort(), true, Duration.ofMillis(200));
- assertTimeoutPreemptively(Duration.ofSeconds(2), () ->
- assertThrows(java.net.http.HttpTimeoutException.class, client::fetchAgentCards));
+ var client =
+ new RegistryClient(
+ "http://127.0.0.1:" + server.getAddress().getPort(), true, Duration.ofMillis(200));
+ assertTimeoutPreemptively(
+ Duration.ofSeconds(2),
+ () -> assertThrows(java.net.http.HttpTimeoutException.class, client::fetchAgentCards));
} finally {
release.countDown();
server.stop(0);