From 784a79554822efca74b75cd9fc51c1519ab5b079 Mon Sep 17 00:00:00 2001 From: Zhoujie628 Date: Wed, 2 Sep 2026 09:05:42 +0800 Subject: [PATCH] fix(engine): harden protocol runtime and workflow validation - deliver Notification-T stream events incrementally and route channels explicitly - enforce AgentCard security requirements and serialize credential refreshes - fail fast on invalid workflow, routing, registry, and orchestration responses - centralize transport-close classification and sensitive-data redaction - add regression coverage for protocol, authentication, and model boundaries Signed-off-by: Zhoujie628 --- .../engine/client/A2AJavaClientRuntime.java | 2 + .../workflow/engine/client/A2ATransport.java | 26 ++-- .../engine/client/AgentCardJacksonModule.java | 32 ++++- .../engine/client/AgentCredentialService.java | 40 +++--- .../client/CredentialHeaderContributor.java | 61 +++++++--- .../client/CredentialHttpTransport.java | 2 +- .../client/DefaultA2AJavaClientRuntime.java | 15 +-- .../client/DefaultWorkflowEngineClient.java | 42 +------ .../engine/client/JdkHttpClientFactory.java | 2 +- .../engine/client/ProtocolResponses.java | 20 ++- .../engine/client/TransportFailures.java | 35 ++++++ .../workflow/engine/client/WireLog.java | 87 +------------ .../engine/client/WorkflowEngineClient.java | 6 +- .../engine/core/ProtocolResultAdapter.java | 14 ++- .../engine/core/WorkflowExecutor.java | 36 +++--- .../workflow/engine/model/StepType.java | 2 +- .../workflow/engine/model/TaskResult.java | 5 +- .../workflow/engine/model/Workflow.java | 76 ++++++++---- .../workflow/engine/registry/LoadPsop.java | 17 ++- .../engine/registry/RegistryClient.java | 37 +++--- .../engine/util/SensitiveDataRedactor.java | 105 ++++++++++++++++ .../client/AgentCardNormalizerTest.java | 35 ++++++ .../client/AgentCredentialServiceTest.java | 84 +++++++++++++ .../CredentialHeaderContributorTest.java | 115 ++++++++++++++++++ .../client/JdkHttpClientFactoryTest.java | 22 ++++ .../engine/client/ProtocolResponsesTest.java | 23 ++++ .../client/RuntimeChannelSelectionTest.java | 30 +++++ .../engine/client/TransportFailuresTest.java | 24 ++++ .../core/ProtocolResultAdapterTest.java | 18 +++ .../engine/core/WorkflowExecutorTest.java | 82 +++++++++++++ .../engine/model/WorkflowModelTest.java | 35 ++++-- .../engine/registry/LoadPsopSearchTest.java | 13 +- .../engine/registry/RegistryClientTest.java | 30 +++++ 33 files changed, 911 insertions(+), 262 deletions(-) create mode 100644 workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java create mode 100644 workflow-engine/src/main/java/dev/openan/workflow/engine/util/SensitiveDataRedactor.java create mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java create mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/JdkHttpClientFactoryTest.java create mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/RuntimeChannelSelectionTest.java create mode 100644 workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java index d352e9eb..49b9f747 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/A2AJavaClientRuntime.java @@ -41,6 +41,8 @@ * engine package so the engine can depend on it directly. */ public interface A2AJavaClientRuntime { + String CHANNEL_STATE_KEY = "dev.openan.workflow.engine.channel"; + String NOTIFICATION_CHANNEL = "notification"; /** * Send a message to an agent and collect the streaming events. 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 fc9eb8ab..ba58ba5d 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 @@ -380,17 +380,6 @@ static boolean isAcknowledgementState(String state) { return state != null && !state.isBlank(); } - private static boolean causedByInterruption(Throwable error) { - Throwable current = error; - while (current != null) { - if (current instanceof InterruptedException) { - return true; - } - current = current.getCause(); - } - return false; - } - private static Map taskTrace(String agentName, String taskId) { Map trace = new HashMap<>(WireLog.context()); trace.put("agent", agentName); @@ -582,6 +571,11 @@ public NotificationSubscription openNotificationStream( MessageSendParams params = buildMessageSendParams(content, contextId, null); ClientCallContext callContext = buildContentCallContext(agentCard, agentName, content); + callContext + .getState() + .put( + A2AJavaClientRuntime.CHANNEL_STATE_KEY, + A2AJavaClientRuntime.NOTIFICATION_CHANNEL); String endpoint = agentCard.supportedInterfaces().isEmpty() ? "?" @@ -602,8 +596,8 @@ public NotificationSubscription openNotificationStream( event.getClass().getSimpleName()); subscription.recordEvent(); if (eventSink != null) { - responses.accept(event); - for (ReceivedMessage received : responses.snapshots()) { + for (ReceivedMessage received : + responses.acceptIncrementally(event)) { eventSink.accept(subscription, received); } } @@ -629,12 +623,8 @@ public NotificationSubscription openNotificationStream( } subscription.completeStream(); } catch (Exception e) { - String msg = e.getMessage() != null ? e.getMessage() : ""; boolean connectionClosed = - !subscription.isActive() - || causedByInterruption(e) - || msg.contains("connection closed locally") - || msg.contains("chunked transfer encoding, state: READING_LENGTH"); + !subscription.isActive() || TransportFailures.isExpectedLocalClose(e); if (connectionClosed) { log.info("[Transport] Notification-T stream closed for {}", agentName); } else { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCardJacksonModule.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCardJacksonModule.java index 79474328..bc499deb 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCardJacksonModule.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCardJacksonModule.java @@ -101,11 +101,15 @@ yield new HTTPAuthSecurityScheme( throw ctxt.instantiationException( APIKeySecurityScheme.class, "apiKeySecurityScheme: 'name' field is required"); } + String location = textOrNull(inner, "location"); + if (location == null) location = textOrNull(inner, "in"); + if (location == null) { + throw ctxt.instantiationException( + APIKeySecurityScheme.class, + "apiKeySecurityScheme: 'location' field is required"); + } yield new APIKeySecurityScheme( - inner.has("in") && inner.get("in").isTextual() - ? APIKeySecurityScheme.Location.valueOf( - inner.get("in").asText().toUpperCase()) - : null, + APIKeySecurityScheme.Location.fromString(location), name, textOrNull(inner, "description")); } @@ -149,7 +153,25 @@ public org.a2aproject.sdk.spec.SecurityRequirement deserialize( var schemes = new java.util.LinkedHashMap>(); if (node.has("schemes") && node.get("schemes").isObject()) { for (var entry : node.get("schemes").properties()) { - schemes.put(entry.getKey(), java.util.List.of()); + JsonNode scopes = entry.getValue(); + if (scopes.isArray()) { + var values = new java.util.ArrayList(); + for (JsonNode scope : scopes) { + if (!scope.isTextual()) { + throw ctxt.weirdStringException( + scope.toString(), String.class, "security requirement scopes must be strings"); + } + values.add(scope.asText()); + } + schemes.put(entry.getKey(), java.util.List.copyOf(values)); + } else if (scopes.isObject() && scopes.isEmpty()) { + schemes.put(entry.getKey(), java.util.List.of()); + } else { + throw ctxt.weirdStringException( + scopes.toString(), + String.class, + "security requirement '" + entry.getKey() + "' scopes must be an array"); + } } } return new org.a2aproject.sdk.spec.SecurityRequirement(schemes); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCredentialService.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCredentialService.java index e11c33a6..6f8f8c6c 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCredentialService.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCredentialService.java @@ -61,6 +61,8 @@ class AgentCredentialService implements CredentialService { private final Map tokenCache = new ConcurrentHashMap<>(); + private final Map refreshLocks = new ConcurrentHashMap<>(); + public AgentCredentialService(String agentName, Map> schemeConfigs) { this(agentName, schemeConfigs, null, null); } @@ -180,23 +182,29 @@ public String getCredential(@NotNull String securitySchemeName, ClientCallContex return null; } - // Check cache TokenEntry cached = tokenCache.get(securitySchemeName); if (cached != null && !cached.isExpired()) { log.info("[Auth] Cache hit for agent {} scheme {}", agentName, securitySchemeName); return cached.token; } - // Login and cache - String token = login(schemeCfg); - if (token != null) { - long ttl = toLong(schemeCfg.get("token_ttl")); - tokenCache.put( - securitySchemeName, new TokenEntry(token, System.currentTimeMillis() / 1000 + ttl)); - log.info("[Auth] Login succeeded: agent={}, scheme={}", agentName, securitySchemeName); - } + Object refreshLock = refreshLocks.computeIfAbsent(securitySchemeName, ignored -> new Object()); + synchronized (refreshLock) { + cached = tokenCache.get(securitySchemeName); + if (cached != null && !cached.isExpired()) { + log.info("[Auth] Cache hit for agent {} scheme {}", agentName, securitySchemeName); + return cached.token; + } - return token; + String token = login(schemeCfg); + if (token != null) { + long ttl = toLong(schemeCfg.get("token_ttl")); + tokenCache.put( + securitySchemeName, new TokenEntry(token, System.currentTimeMillis() / 1000 + ttl)); + log.info("[Auth] Login succeeded: agent={}, scheme={}", agentName, securitySchemeName); + } + return token; + } } private String login(Map schemeCfg) { @@ -245,6 +253,10 @@ private String login(Map schemeCfg) { .build(); HttpResponse resp = httpClient.send(request, HttpResponse.BodyHandlers.ofString()); return parseTokenResponse(tokenField, resp.statusCode(), resp.body()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.error("[Auth] Login interrupted: agent={}, url={}", agentName, loginUrl); + return null; } catch (Exception e) { log.error( "[Auth] Login failed: agent={}, url={}, error={}", agentName, loginUrl, e.getMessage()); @@ -262,10 +274,10 @@ private String parseTokenResponse(String tokenField, int status, String response mapper.readValue(responseBody, new TypeReference>() {}); String token = extractNestedValue(data, tokenField); if (token == null) { - token = - (String) - data.getOrDefault( - "accessSession", data.getOrDefault("access_token", data.get("token"))); + log.error( + "[Auth] Login response for agent {} does not contain configured token field {}", + agentName, + tokenField); } return token; } 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 760b3751..460f5b85 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 @@ -30,22 +30,27 @@ final class CredentialHeaderContributor implements HeaderContributor { } private static void addCredentialHeader( - String agentName, 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", ""); - headers.put(authHeader, prefix + credential); - log.info("[Auth] Set header {} for agent {}", authHeader, agentName); + putHeader(headers, authHeader, prefix + credential); } else { - headers.put("Authorization", "Bearer " + credential); - log.info("[Auth] Set Bearer header for agent {}", agentName); + putHeader(headers, "Authorization", "Bearer " + credential); } String acceptHeader = (String) schemeConfig.get("accept_header"); if (acceptHeader != null && !acceptHeader.isEmpty()) { - headers.put("Accept", acceptHeader); + putHeader(headers, "Accept", acceptHeader); + } + } + + private static void putHeader(Map headers, String name, String value) { + String existing = headers.putIfAbsent(name, value); + if (existing != null && !existing.equals(value)) { + throw new SecurityException( + "Authentication schemes require conflicting values for header " + name); } } @@ -84,30 +89,46 @@ public Map contribute( "Authentication failed for agent " + agentName + ": credential configuration is empty"); } + java.util.ArrayList failures = new java.util.ArrayList<>(); for (SecurityRequirement requirement : securityRequirements) { + Map candidate = new HashMap<>(); Map> schemes = requirement.schemes(); + String failure = null; for (String schemeName : schemes.keySet()) { + if (!securitySchemes.containsKey(schemeName)) { + failure = "scheme " + schemeName + " is not declared by the AgentCard"; + break; + } Map schemeConfig = schemeConfigs.get(schemeName); if (schemeConfig == null) { - throw new SecurityException( - "Authentication failed for agent " - + agentName - + ": no configuration for scheme " - + schemeName); + failure = "no configuration for scheme " + schemeName; + break; } String credential = credentialService.getCredential(schemeName, null); if (credential == null) { - throw new SecurityException( - "Authentication failed for agent " - + agentName - + " (scheme=" - + schemeName - + "), request blocked"); + failure = "no credential for scheme " + schemeName; + break; } - addCredentialHeader(agentName, headers, schemeConfig, credential); - return headers; + try { + addCredentialHeader(candidate, schemeConfig, credential); + } catch (SecurityException error) { + failure = error.getMessage(); + break; + } + } + if (failure == null) { + log.info( + "[Auth] Satisfied security requirement for agent {}: {}", agentName, schemes.keySet()); + return candidate; } + failures.add(failure); } - return headers; + if (authProvider != null) return headers; + throw new SecurityException( + "Authentication failed for agent " + + agentName + + ": no security requirement can be satisfied (" + + String.join("; ", failures) + + ")"); } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHttpTransport.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHttpTransport.java index 6e9d9e2f..5ae86983 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHttpTransport.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/CredentialHttpTransport.java @@ -46,7 +46,7 @@ Response send(URI uri, String method, String contentType, String body) throws Ex try { connection.setConnectTimeout(timeoutMillis); connection.setReadTimeout(timeoutMillis); - connection.setInstanceFollowRedirects(true); + connection.setInstanceFollowRedirects(false); connection.setRequestMethod(method); connection.setDoOutput(true); connection.setRequestProperty("Content-Type", contentType); diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java index 14faba10..4abc929e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/DefaultA2AJavaClientRuntime.java @@ -217,10 +217,7 @@ private static void onError( } else { String msg = error.getMessage() != null ? error.getMessage() : ""; RemoteProblemException problem = RemoteProblemException.findIn(error); - boolean connectionClosed = - problem == null - && (msg.contains("connection closed locally") - || msg.contains("chunked transfer encoding, state: READING_LENGTH")); + boolean connectionClosed = problem == null && TransportFailures.isExpectedLocalClose(error); if (connectionClosed) { log.debug("[A2ARuntime] Connection closed for '{}': {}", agentName, msg); } else { @@ -311,10 +308,10 @@ private static String describeTerminalEvent(ClientEvent event) { return event.getClass().getSimpleName(); } - private static boolean isNotificationStream(org.a2aproject.sdk.spec.MessageSendParams params) { - return params.message() != null - && params.message().metadata() != null - && params.message().metadata().containsKey(A2ATExtension.NOTIFICATION_T.uri()); + static boolean isNotificationStream(ClientCallContext callContext) { + return callContext != null + && A2AJavaClientRuntime.NOTIFICATION_CHANNEL.equals( + callContext.getState().get(A2AJavaClientRuntime.CHANNEL_STATE_KEY)); } @Override @@ -327,7 +324,7 @@ public Iterable sendMessage( if (closed.get()) throw new IllegalStateException("A2A client runtime is closed"); String agentUrl = extractAgentUrl(agentCard); Client client = - isNotificationStream(params) + isNotificationStream(callContext) ? getOrCreateStreamClient( agentCard, params.message() != null ? params.message().contextId() : null) : getOrCreateClient(agentCard, agentUrl); 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 be70bcd6..305bdedb 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 @@ -36,7 +36,6 @@ import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult; import org.a2aproject.sdk.spec.AgentCard; import org.a2aproject.sdk.spec.ListTasksParams; -import org.a2aproject.sdk.spec.Message; import org.a2aproject.sdk.spec.TaskStatusUpdateEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,8 +51,8 @@ public class DefaultWorkflowEngineClient implements WorkflowEngineClient, AutoCl private final boolean closeTransportOnClose; private final AtomicBoolean closed = new AtomicBoolean(); private final Set invocations = java.util.concurrent.ConcurrentHashMap.newKeySet(); - private EventCallback eventCallback = new EventCallback(); - private ControlPoint controlPoint; + private volatile EventCallback eventCallback = new EventCallback(); + private volatile ControlPoint controlPoint; /** Creates a caller-owned transport facade. */ public DefaultWorkflowEngineClient(A2ATransport transport) { @@ -372,18 +371,10 @@ private void emitAgentResponse(String agentName, SendMessageResult result) { } private void forwardIntermediateEvent(ClientEvent event, String agentName) { + Map data = ClientEventMapper.toMap(event, agentName); if (event instanceof TaskUpdateEvent tue) { if (tue.getUpdateEvent() instanceof TaskStatusUpdateEvent sue) { String state = sue.status().state().name(); - StringBuilder statusText = new StringBuilder(); - A2ATransport.extractTextFromMessage(sue.status().message(), statusText); - Map data = new HashMap<>(); - data.put("agent", agentName); - data.put("state", state); - data.put("is_final", sue.isFinal()); - if (!statusText.isEmpty()) data.put("text", statusText.toString()); - if (sue.metadata() != null && !sue.metadata().isEmpty()) - data.put("metadata", sue.metadata()); log.info( "[EngineClient] Agent {} status update: {} (final={})", agentName, @@ -392,20 +383,6 @@ private void forwardIntermediateEvent(ClientEvent event, String agentName) { emit(EventType.AGENT_STATUS_UPDATE, data); } else if (tue.getUpdateEvent() instanceof org.a2aproject.sdk.spec.TaskArtifactUpdateEvent ae) { - StringBuilder text = new StringBuilder(); - A2ATransport.extractTextFromArtifact(ae.artifact(), text); - Map data = new HashMap<>(); - data.put("agent", agentName); - data.put("artifact_id", ae.artifact().artifactId()); - data.put("artifact_name", ae.artifact().name()); - data.put("append", ae.append()); - data.put("last_chunk", ae.lastChunk()); - if (!text.isEmpty()) data.put("text", text.toString()); - // Business metadata belongs to the Artifact. Event metadata describes delivery - // (chunking, tracing, etc.) and must not hide the protocol payload. - Map artifactMetadata = ae.artifact().metadata(); - if (artifactMetadata != null && !artifactMetadata.isEmpty()) - data.put("metadata", artifactMetadata); log.info( "[EngineClient] Agent {} artifact update: {} ({})", agentName, @@ -413,17 +390,8 @@ private void forwardIntermediateEvent(ClientEvent event, String agentName) { ae.artifact().artifactId()); emit(EventType.AGENT_ARTIFACT_UPDATE, data); } - } else if (event instanceof MessageEvent me) { - Message msg = me.getMessage(); - StringBuilder text = new StringBuilder(); - A2ATransport.extractTextFromMessage(msg, text); - Map data = new HashMap<>(); - data.put("agent", agentName); - data.put("role", msg.role().name()); - if (!text.isEmpty()) data.put("text", text.toString()); - if (msg.metadata() != null && !msg.metadata().isEmpty()) { - data.put("metadata", msg.metadata()); - } + } else if (event instanceof MessageEvent) { + String text = (String) data.getOrDefault("text", ""); log.info("[EngineClient] Agent {} message event: {} chars", agentName, text.length()); emit(EventType.AGENT_MESSAGE_EVENT, data); } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/JdkHttpClientFactory.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/JdkHttpClientFactory.java index 1ad93bdc..4e1e9cec 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/JdkHttpClientFactory.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/JdkHttpClientFactory.java @@ -29,7 +29,7 @@ static HttpClient create( HttpClient.newBuilder() .version(HttpClient.Version.HTTP_1_1) .connectTimeout(connectTimeout) - .followRedirects(HttpClient.Redirect.ALWAYS); + .followRedirects(HttpClient.Redirect.NEVER); if (executor != null) builder.executor(executor); SslContextFactory.create( sslVerify, caCertsPath, clientCertPath, clientKeyPath, clientKeyPassword, crlPath) diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProtocolResponses.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProtocolResponses.java index b679853c..aa936593 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProtocolResponses.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/ProtocolResponses.java @@ -84,6 +84,18 @@ void accept(ClientEvent event) { } } + List acceptIncrementally(ClientEvent event) { + if (event instanceof MessageEvent message) { + Message value = message.getMessage(); + return List.of(new ReceivedMessage(content(value), Map.of(), List.of())); + } + accept(event); + if (event instanceof TaskEvent || event instanceof TaskUpdateEvent) { + return List.of(taskSnapshot()); + } + return List.of(); + } + private void snapshot(Task task, boolean authoritative) { if (task == null) return; hasTask = true; @@ -121,10 +133,12 @@ private void append(TaskArtifactUpdateEvent event) { List snapshots() { List result = new ArrayList<>(); messages.values().forEach(m -> result.add(new ReceivedMessage(m, Map.of(), List.of()))); - if (hasTask) - result.add( - new ReceivedMessage(statusMessage, taskMetadata, new ArrayList<>(artifacts.values()))); + if (hasTask) result.add(taskSnapshot()); return List.copyOf(result); } + + private ReceivedMessage taskSnapshot() { + return new ReceivedMessage(statusMessage, taskMetadata, new ArrayList<>(artifacts.values())); + } } } 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 new file mode 100644 index 00000000..6bf0f1de --- /dev/null +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/TransportFailures.java @@ -0,0 +1,35 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.client; + +import java.nio.channels.AsynchronousCloseException; +import java.nio.channels.ClosedChannelException; +import java.util.Locale; +import java.util.concurrent.CancellationException; + +/** Classifies transport termination without making provider-specific messages the primary signal. */ +final class TransportFailures { + private TransportFailures() {} + + static boolean isExpectedLocalClose(Throwable error) { + for (Throwable current = error; current != null; current = current.getCause()) { + if (current instanceof InterruptedException + || current instanceof CancellationException + || current instanceof AsynchronousCloseException + || current instanceof ClosedChannelException) { + return true; + } + String message = current.getMessage(); + if (message != null) { + String normalized = message.toLowerCase(Locale.ROOT); + if (normalized.contains("connection closed locally") + || normalized.contains("chunked transfer encoding, state: reading_length")) { + return true; + } + } + } + return false; + } +} diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WireLog.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WireLog.java index 48a8e5e8..13c1c680 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WireLog.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WireLog.java @@ -19,25 +19,18 @@ package dev.openan.workflow.engine.client; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; import java.io.ByteArrayOutputStream; import java.nio.ByteBuffer; import java.nio.charset.StandardCharsets; import java.util.*; import java.util.function.Consumer; -import java.util.regex.Pattern; import org.slf4j.LoggerFactory; /** Application-boundary evidence, not a TLS/HTTP packet capture. Never logs secrets verbatim. */ public final class WireLog { private static final org.slf4j.Logger LOG = LoggerFactory.getLogger("PROTOCOL"); private static final ThreadLocal> CONTEXT = ThreadLocal.withInitial(Map::of); - private static final com.fasterxml.jackson.databind.ObjectMapper JSON = - new com.fasterxml.jackson.databind.ObjectMapper(); - private static final Pattern SECRET_FORM = - Pattern.compile( - "(?i)((?:password|passwd|pwd|[\\w-]*token|[\\w-]*secret|accessSession|api[-_]?key)=)[^&\\s]*"); - private static final Pattern AUTH_VALUE = - Pattern.compile("(?i)(\\b(?:Bearer|Basic)[ \\t]+)[^\\s\"\\\\,;]+"); private WireLog() {} @@ -85,87 +78,15 @@ static int limit() { } public static String redact(String text) { - if (text == null) return ""; - StringBuilder result = new StringBuilder(); - int copied = 0; - for (int cursor = 0; cursor < text.length(); cursor++) { - if (text.charAt(cursor) != '"') continue; - int endOfString = valueEnd(text, cursor); - if (endOfString <= cursor || text.charAt(endOfString - 1) != '"') break; - int colon = endOfString; - while (colon < text.length() && Character.isWhitespace(text.charAt(colon))) colon++; - if (colon == text.length() || text.charAt(colon) != ':') { - cursor = endOfString - 1; - continue; - } - String key; - try { - key = JSON.readValue(text.substring(cursor, endOfString), String.class); - } catch (com.fasterxml.jackson.core.JsonProcessingException invalid) { - cursor = endOfString - 1; - continue; - } - cursor = colon; - if (!sensitive(key)) continue; - int start = colon + 1; - while (start < text.length() && Character.isWhitespace(text.charAt(start))) start++; - int end = valueEnd(text, start); - result.append(text, copied, start).append("\"***\""); - copied = end; - cursor = end - 1; - } - result.append(text, copied, text.length()); - return AUTH_VALUE.matcher(SECRET_FORM.matcher(result).replaceAll("$1***")).replaceAll("$1***"); - } - - /** Skip a complete JSON value without reserializing the surrounding wire text. */ - private static int valueEnd(String text, int start) { - boolean quoted = false; - boolean escaped = false; - int depth = 0; - for (int i = start; i < text.length(); i++) { - char c = text.charAt(i); - if (quoted) { - if (escaped) escaped = false; - else if (c == '\\') escaped = true; - else if (c == '"') { - quoted = false; - if (depth == 0) return i + 1; - } - } else if (c == '"') quoted = true; - else if (c == '{' || c == '[') depth++; - else if (c == '}' || c == ']') { - if (depth == 0) return i; - if (--depth == 0) return i + 1; - } else if (depth == 0 && (c == ',' || Character.isWhitespace(c))) return i; - } - // Incomplete secret values are redacted to the end, never logged as a raw prefix. - return text.length(); + return SensitiveDataRedactor.redact(text); } static boolean sensitive(String name) { - String key = name.toLowerCase(Locale.ROOT); - // URI-shaped keys are protocol identifiers (e.g. the Authorization-T extension URI in message - // metadata), not credential fields; their values are business content and must stay visible. - if (key.contains("://")) return false; - return key.contains("authorization") - || key.contains("cookie") - || key.contains("token") - || key.contains("secret") - || key.contains("password") - || key.equals("pwd") - || key.equals("passwd") - || key.equals("accesssession") - || key.equals("x-api-key") - || key.equals("api-key") - || key.equals("apikey"); + return SensitiveDataRedactor.sensitive(name); } public static Map> safeHeaders(Map> headers) { - Map> result = new LinkedHashMap<>(); - headers.forEach( - (key, values) -> result.put(key, sensitive(key) ? List.of("***") : List.copyOf(values))); - return Collections.unmodifiableMap(result); + return SensitiveDataRedactor.safeHeaders(headers); } /** Actual serialized data only. SDK-decoded data must explicitly say decoded-data. */ diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java index a22b65bf..20e4d06b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClient.java @@ -39,7 +39,11 @@ CompletableFuture dispatch( /** Sends final content outside the DAG using explicitly configured interaction callbacks. */ CompletableFuture sendMessage(String agentName, MessageContent content); - /** Maximum wait for each business callback, including initial content preparation. */ + /** + * Maximum wait for one task interaction, from initial content preparation through remote task + * completion and any negotiation exchanges. The same budget applies independently to a route + * callback. + */ default long callbackTimeoutSeconds() { return 600; } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/ProtocolResultAdapter.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/ProtocolResultAdapter.java index 632f3231..d5b2c273 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/ProtocolResultAdapter.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/ProtocolResultAdapter.java @@ -21,6 +21,7 @@ import dev.openan.workflow.engine.model.SendMessageResult; import dev.openan.workflow.engine.model.TaskResult; +import org.a2aproject.sdk.spec.TaskState; /** Maps transport state independently from the presence or absence of business content. */ final class ProtocolResultAdapter { @@ -30,10 +31,19 @@ static TaskResult toTaskResult(SendMessageResult result) { if (result == null) return TaskResult.failure("remote.no_response", "Agent returned no response"); String state = result.getTaskState(); + TaskState protocolState = null; + if (state != null && !state.isBlank()) { + try { + protocolState = TaskState.valueOf(state); + } catch (IllegalArgumentException ignored) { + // Unknown transport state remains a failure and is preserved in the diagnostic below. + } + } boolean standaloneMessage = result.getTask() == null && !result.getReceivedMessages().isEmpty(); boolean success = - "TASK_STATE_COMPLETED".equals(state) - || ((state == null || state.isBlank() || state.endsWith("UNSPECIFIED")) + protocolState == TaskState.TASK_STATE_COMPLETED + || (((state == null || state.isBlank()) + || protocolState == TaskState.TASK_STATE_UNSPECIFIED) && standaloneMessage); if (result.getFailureCode() != null) success = false; return TaskResult.builder() diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java index 9ea2cff5..def3deaa 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/core/WorkflowExecutor.java @@ -34,6 +34,7 @@ import dev.openan.workflow.engine.model.Workflow; import dev.openan.workflow.engine.model.WorkflowInput; import dev.openan.workflow.engine.model.WorkflowStep; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; import java.util.ArrayList; import java.util.Collections; import java.util.Deque; @@ -56,6 +57,9 @@ */ public class WorkflowExecutor { private static final Logger log = LoggerFactory.getLogger(WorkflowExecutor.class); + private static final Set TERMINAL_ROUTES = Set.of("end", "retry", "endNode"); + private static final com.fasterxml.jackson.databind.ObjectMapper WORKFLOW_SNAPSHOT_MAPPER = + new com.fasterxml.jackson.databind.ObjectMapper(); private final Workflow workflow; private final AtomicBoolean started = new AtomicBoolean(); @@ -80,23 +84,16 @@ public WorkflowExecutor( EventCallback eventCallback, String runtimeIntent, String lang) { - this.workflow = - new com.fasterxml.jackson.databind.ObjectMapper().convertValue(workflow, Workflow.class); + // Definitions are mutable for JSON/builder compatibility. Snapshot them so runtime status + // changes and caller mutations cannot leak between concurrent executions. + this.workflow = WORKFLOW_SNAPSHOT_MAPPER.convertValue(workflow, Workflow.class); this.controlPoint = controlPoint; this.engineClient = engineClient; this.eventCallback = eventCallback != null ? eventCallback : new EventCallback(); this.contextBuilder = new ContextBuilder(this.workflow, runtimeIntent); this.lang = lang != null ? lang : "zh"; - try { - this.engineClient.setControlPoint(this.controlPoint); - } catch (Exception ignored) { - // Engine client may not support control point injection - } - try { - this.engineClient.setEventCallback(this.eventCallback); - } catch (Exception ignored) { - // Engine client may not support event callback injection - } + this.engineClient.setControlPoint(this.controlPoint); + this.engineClient.setEventCallback(this.eventCallback); log.info( "[Executor] Workflow: {}, steps={}, intent={}, lang={}", workflow.getName(), @@ -106,7 +103,7 @@ public WorkflowExecutor( } private static boolean isTerminalRoute(String stepName) { - return "end".equals(stepName) || "retry".equals(stepName) || "endNode".equals(stepName); + return TERMINAL_ROUTES.contains(stepName); } private static void completeFrom( @@ -387,6 +384,10 @@ private void validateWorkflowGraph() { if (name == null || name.isBlank()) { throw new IllegalArgumentException("Workflow step name must not be blank"); } + if (isTerminalRoute(name)) { + throw new IllegalArgumentException( + "Workflow step name is reserved for a terminal route: " + name); + } if (indices.put(name, i) != null) { throw new IllegalArgumentException("Duplicate workflow step name: " + name); } @@ -605,7 +606,7 @@ private StepResult processTaskResult( taskId(step.getName(), subtaskIndex), task.getAgent(), response.getErrorCode(), - dev.openan.workflow.engine.client.WireLog.redact(response.getError()) + SensitiveDataRedactor.redact(response.getError()) .replace("\r", "\\r") .replace("\n", "\\n")); } @@ -802,11 +803,14 @@ private CompletableFuture> determineNextSteps( new dev.openan.workflow.engine.model.RouteRequest.RouteOption( option.getStep(), option.getCondition())) .toList()); - return controlPoint - .onRoute(routeRequest) + CompletableFuture routeDecision = + java.util.Objects.requireNonNull( + controlPoint.onRoute(routeRequest), "onRoute returned null future"); + return routeDecision .orTimeout(engineClient.callbackTimeoutSeconds(), java.util.concurrent.TimeUnit.SECONDS) .thenApply( decision -> { + java.util.Objects.requireNonNull(decision, "onRoute returned null decision"); log.info( "Route for '{}': {} ({})", step.getName(), diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/StepType.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/StepType.java index 0b60e682..006a6c39 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/StepType.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/StepType.java @@ -39,7 +39,7 @@ public static StepType fromValue(String v) { return t; } } - return ALL_SUCCESS; + throw new IllegalArgumentException("Unknown workflow step type: " + v); } public String getValue() { 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 fb9fef30..039ead65 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 @@ -32,7 +32,10 @@ public class TaskResult { /** Ordered business outputs; values may be text or JSON-serializable structured data. */ @Builder.Default List outputs = List.of(); - /** Remote evidence; local onSelfTask outputs leave this empty. */ + /** + * Remote evidence; local onSelfTask outputs leave this empty. When present, {@code outputs} are + * derived from these messages so conflicting builder-supplied output values are not retained. + */ @Builder.Default List receivedMessages = List.of(); String error; diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Workflow.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Workflow.java index bd7e01b4..da7503ba 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Workflow.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/Workflow.java @@ -39,30 +39,28 @@ public class Workflow { @SuppressWarnings("unchecked") public static Workflow fromMap(Map data) { + if (data == null) throw new IllegalArgumentException("Workflow data must not be null"); Workflow wf = new Workflow(); - wf.setId((String) data.getOrDefault("id", "")); - wf.setName((String) data.getOrDefault("name", "")); - wf.setDescription((String) data.getOrDefault("description", "")); - List steps = - parseSteps((List>) data.getOrDefault("steps", List.of())); + wf.setId(stringValue(data, "id", "")); + wf.setName(stringValue(data, "name", "")); + wf.setDescription(stringValue(data, "description", "")); + List steps = parseSteps(mapList(data.get("steps"), "steps")); wf.setSteps(steps); return wf; } - @SuppressWarnings("unchecked") private static List parseSteps(List> stepList) { List steps = new ArrayList<>(); for (Map s : stepList) { - List subtasks = - parseSubtasks((List>) s.getOrDefault("subtasks", List.of())); - List nextList = - parseNextSteps((List>) s.getOrDefault("next", List.of())); + List subtasks = parseSubtasks(mapList(s.get("subtasks"), "steps[].subtasks")); + List nextList = parseNextSteps(mapList(s.get("next"), "steps[].next")); List contextFrom = parseContextFrom(s.get("context_from")); int layer = s.get("layer") instanceof Number num ? num.intValue() : 0; - String stValue = (String) s.getOrDefault("step_type", s.getOrDefault("type", "AllSuccess")); + Object rawStepType = s.containsKey("step_type") ? s.get("step_type") : s.get("type"); + String stValue = rawStepType == null ? "AllSuccess" : requireString(rawStepType, "step_type"); steps.add( WorkflowStep.builder() - .name((String) s.getOrDefault("name", "")) + .name(stringValue(s, "name", "")) .subtasks(subtasks) .next(nextList) .layer(layer) @@ -73,37 +71,73 @@ private static List parseSteps(List> stepList) return steps; } - @SuppressWarnings("unchecked") private static List parseSubtasks(List> stList) { List subtasks = new ArrayList<>(); for (Map t : stList) { subtasks.add( Task.builder() - .agent((String) t.getOrDefault("agent", "")) - .skill((String) t.getOrDefault("skill", "")) - .description((String) t.getOrDefault("description", "")) + .agent(stringValue(t, "agent", "")) + .skill(stringValue(t, "skill", "")) + .description(stringValue(t, "description", "")) .build()); } return subtasks; } - @SuppressWarnings("unchecked") private static List parseNextSteps(List> jcList) { List nextList = new ArrayList<>(); for (Map jc : jcList) { nextList.add( JumpCondition.builder() - .step((String) jc.getOrDefault("step", "")) - .condition((String) jc.getOrDefault("condition", "")) + .step(stringValue(jc, "step", "")) + .condition(stringValue(jc, "condition", "")) .build()); } return nextList; } - @SuppressWarnings("unchecked") private static List parseContextFrom(Object cfRaw) { - if (cfRaw instanceof List) return (List) cfRaw; + if (cfRaw instanceof List values) { + List result = new ArrayList<>(values.size()); + for (Object value : values) result.add(requireString(value, "context_from[]")); + return List.copyOf(result); + } if (cfRaw instanceof String cfStr && !cfStr.isEmpty()) return List.of(cfStr); + if (cfRaw != null && !(cfRaw instanceof String)) { + throw new IllegalArgumentException("context_from must be a string or a list of strings"); + } return null; } + + private static String stringValue(Map data, String key, String fallback) { + Object value = data.get(key); + return value == null ? fallback : requireString(value, key); + } + + private static String requireString(Object value, String field) { + if (value instanceof String text) return text; + throw new IllegalArgumentException(field + " must be a string"); + } + + private static List> mapList(Object value, String field) { + if (value == null) return List.of(); + if (!(value instanceof List values)) { + throw new IllegalArgumentException(field + " must be a list"); + } + List> result = new ArrayList<>(values.size()); + for (Object item : values) { + if (!(item instanceof Map raw)) { + throw new IllegalArgumentException(field + " entries must be objects"); + } + Map entry = new java.util.LinkedHashMap<>(); + for (Map.Entry fieldEntry : raw.entrySet()) { + if (!(fieldEntry.getKey() instanceof String key)) { + throw new IllegalArgumentException(field + " object keys must be strings"); + } + entry.put(key, fieldEntry.getValue()); + } + result.add(java.util.Collections.unmodifiableMap(entry)); + } + return List.copyOf(result); + } } 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 d798d0a3..6c33ce7f 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 @@ -23,6 +23,7 @@ import dev.openan.workflow.engine.client.SslContextFactory; import dev.openan.workflow.engine.model.Workflow; import dev.openan.workflow.engine.model.WorkflowSearchResult; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; import java.io.InputStream; import java.io.OutputStream; import java.net.HttpURLConnection; @@ -64,7 +65,7 @@ public static Workflow load(String baseUrl, String psopId, String accessToken, b log.info("[Registry] Loading PSOP from {} (ssl_verify={})", anonymousUrl(url, accessToken), sslVerify); HttpResult resp = execute("GET", url, null, sslVerify); if (resp.statusCode() != 200) { - throw new RuntimeException("Orchestration center returned " + resp.statusCode()); + throw requestFailure(resp); } @SuppressWarnings("unchecked") Map data = mapper.readValue(resp.body(), Map.class); @@ -93,7 +94,7 @@ public static List search( String jsonBody = mapper.writeValueAsString(Map.of("intent", intent, "top_n", topN)); HttpResult resp = execute("POST", url, jsonBody, sslVerify); if (resp.statusCode() != 200) { - throw new RuntimeException("Orchestration center returned " + resp.statusCode()); + throw requestFailure(resp); } @SuppressWarnings("unchecked") Map data = mapper.readValue(resp.body(), Map.class); @@ -126,6 +127,16 @@ static String anonymousUrl(String url, String token) { : url.substring(0, url.indexOf("?access_token=")) + "?access_token="; } + private static RuntimeException requestFailure(HttpResult response) { + String detail = + SensitiveDataRedactor.redact(response.body()).replace("\r", "\\r").replace("\n", "\\n"); + if (detail.length() > 512) detail = detail.substring(0, 512) + "..."; + return new RuntimeException( + "Orchestration center returned " + + response.statusCode() + + (detail.isBlank() ? "" : ": " + detail)); + } + private static HttpResult execute(String method, String url, String jsonBody, boolean sslVerify) throws Exception { HttpURLConnection connection = (HttpURLConnection) URI.create(url).toURL().openConnection(); @@ -140,7 +151,7 @@ private static HttpResult execute(String method, String url, String jsonBody, bo https.getURL().getPort()); } try { - connection.setInstanceFollowRedirects(true); + connection.setInstanceFollowRedirects(false); connection.setConnectTimeout(30_000); connection.setReadTimeout(30_000); connection.setRequestMethod(method); 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 fe29caeb..002546bb 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 @@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import dev.openan.workflow.engine.client.AgentCardNormalizer; import dev.openan.workflow.engine.client.SslContextFactory; +import dev.openan.workflow.engine.util.SensitiveDataRedactor; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -57,13 +58,25 @@ public RegistryClient(String url, boolean sslVerify, java.time.Duration requestT HttpClient.Builder clientBuilder = HttpClient.newBuilder() .connectTimeout(java.time.Duration.ofSeconds(30)) - .followRedirects(HttpClient.Redirect.ALWAYS); + .followRedirects(HttpClient.Redirect.NEVER); if (!sslVerify) { clientBuilder.sslContext(SslContextFactory.createTrustAll()); } this.httpClient = clientBuilder.build(); } + private static RuntimeException requestFailure(String service, int status, String body) { + String detail = SensitiveDataRedactor.redact(body).replace("\r", "\\r").replace("\n", "\\n"); + if (detail.length() > 512) detail = detail.substring(0, 512) + "..."; + return new RuntimeException( + service + " returned " + status + (detail.isBlank() ? "" : ": " + detail)); + } + + @SuppressWarnings("unchecked") + public Map fetchAgentCard(String name) throws Exception { + return fetchAgentCard(name, null); + } + @SuppressWarnings("unchecked") public List> fetchAgentCards() throws Exception { String url = baseUrl + "/rest/v1/registry-center/agent-cards"; @@ -71,7 +84,7 @@ public List> fetchAgentCards() throws Exception { HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build(); HttpResponse resp = send(request); if (resp.statusCode() != 200) { - throw new RuntimeException("Registry returned " + resp.statusCode()); + throw requestFailure("Registry", resp.statusCode(), resp.body()); } Map data = mapper.readValue(resp.body(), Map.class); List> cards = @@ -83,9 +96,8 @@ public List> fetchAgentCards() throws Exception { return cards; } - @SuppressWarnings("unchecked") - public Map fetchAgentCard(String name) throws Exception { - return fetchAgentCard(name, null); + public String getBaseUrl() { + return baseUrl; } @SuppressWarnings("unchecked") @@ -107,7 +119,7 @@ public Map fetchAgentCard(String name, String organization) thro HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).timeout(requestTimeout).GET().build(); HttpResponse resp = send(request); if (resp.statusCode() != 200) { - throw new RuntimeException("Registry returned " + resp.statusCode()); + throw requestFailure("Registry", resp.statusCode(), resp.body()); } Map data = mapper.readValue(resp.body(), Map.class); List> cards = @@ -121,10 +133,6 @@ public Map fetchAgentCard(String name, String organization) thro return AgentCardNormalizer.normalize(cards.get(0)); } - public String getBaseUrl() { - return baseUrl; - } - /** * Register or update an AgentCard in the registry. * @@ -145,12 +153,11 @@ public Map registerAgentCard(Map agentCard) thro .POST(HttpRequest.BodyPublishers.ofString(json)) .build(); HttpResponse resp = send(request); - Map result = mapper.readValue(resp.body(), Map.class); - if (resp.statusCode() == 200 || resp.statusCode() == 201) { - log.info("[Registry] Agent card registered: name={}", agentCard.get("name")); - } else { - log.warn("[Registry] Registration returned {}: {}", resp.statusCode(), resp.body()); + if (resp.statusCode() != 200 && resp.statusCode() != 201) { + throw requestFailure("Registry registration", resp.statusCode(), resp.body()); } + Map result = mapper.readValue(resp.body(), Map.class); + log.info("[Registry] Agent card registered: name={}", agentCard.get("name")); return result; } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/util/SensitiveDataRedactor.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/util/SensitiveDataRedactor.java new file mode 100644 index 00000000..7fe916fc --- /dev/null +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/util/SensitiveDataRedactor.java @@ -0,0 +1,105 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.util; + +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.regex.Pattern; + +/** Redacts credential-shaped values from diagnostics without reserializing business payloads. */ +public final class SensitiveDataRedactor { + private static final com.fasterxml.jackson.databind.ObjectMapper JSON = + new com.fasterxml.jackson.databind.ObjectMapper(); + private static final Pattern SECRET_FORM = + Pattern.compile( + "(?i)((?:password|passwd|pwd|[\\w-]*token|[\\w-]*secret|accessSession|api[-_]?key)=)[^&\\s]*"); + private static final Pattern AUTH_VALUE = + Pattern.compile("(?i)(\\b(?:Bearer|Basic)[ \\t]+)[^\\s\"\\\\,;]+"); + + private SensitiveDataRedactor() {} + + public static String redact(String text) { + if (text == null) return ""; + StringBuilder result = new StringBuilder(); + int copied = 0; + for (int cursor = 0; cursor < text.length(); cursor++) { + if (text.charAt(cursor) != '"') continue; + int endOfString = valueEnd(text, cursor); + if (endOfString <= cursor || text.charAt(endOfString - 1) != '"') break; + int colon = endOfString; + while (colon < text.length() && Character.isWhitespace(text.charAt(colon))) colon++; + if (colon == text.length() || text.charAt(colon) != ':') { + cursor = endOfString - 1; + continue; + } + String key; + try { + key = JSON.readValue(text.substring(cursor, endOfString), String.class); + } catch (com.fasterxml.jackson.core.JsonProcessingException invalid) { + cursor = endOfString - 1; + continue; + } + cursor = colon; + if (!sensitive(key)) continue; + int start = colon + 1; + while (start < text.length() && Character.isWhitespace(text.charAt(start))) start++; + int end = valueEnd(text, start); + result.append(text, copied, start).append("\"***\""); + copied = end; + cursor = end - 1; + } + result.append(text, copied, text.length()); + return AUTH_VALUE.matcher(SECRET_FORM.matcher(result).replaceAll("$1***")).replaceAll("$1***"); + } + + private static int valueEnd(String text, int start) { + boolean quoted = false; + boolean escaped = false; + int depth = 0; + for (int i = start; i < text.length(); i++) { + char c = text.charAt(i); + if (quoted) { + if (escaped) escaped = false; + else if (c == '\\') escaped = true; + else if (c == '"') { + quoted = false; + if (depth == 0) return i + 1; + } + } else if (c == '"') quoted = true; + else if (c == '{' || c == '[') depth++; + else if (c == '}' || c == ']') { + if (depth == 0) return i; + if (--depth == 0) return i + 1; + } else if (depth == 0 && (c == ',' || Character.isWhitespace(c))) return i; + } + return text.length(); + } + + public static boolean sensitive(String name) { + String key = name.toLowerCase(Locale.ROOT); + if (key.contains("://")) return false; + return key.contains("authorization") + || key.contains("cookie") + || key.contains("token") + || key.contains("secret") + || key.contains("password") + || key.equals("pwd") + || key.equals("passwd") + || key.equals("accesssession") + || key.equals("x-api-key") + || key.equals("api-key") + || key.equals("apikey"); + } + + public static Map> safeHeaders(Map> headers) { + Map> result = new LinkedHashMap<>(); + headers.forEach( + (key, values) -> result.put(key, sensitive(key) ? List.of("***") : List.copyOf(values))); + return Collections.unmodifiableMap(result); + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCardNormalizerTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCardNormalizerTest.java index 9801e29b..0594f91f 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCardNormalizerTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCardNormalizerTest.java @@ -21,8 +21,12 @@ import static org.junit.jupiter.api.Assertions.*; +import com.fasterxml.jackson.databind.ObjectMapper; import java.util.List; import java.util.Map; +import org.a2aproject.sdk.spec.APIKeySecurityScheme; +import org.a2aproject.sdk.spec.SecurityRequirement; +import org.a2aproject.sdk.spec.SecurityScheme; import org.junit.jupiter.api.Test; /** Tests for AgentCardNormalizer: OpenAPI -> structured format conversion. */ @@ -79,6 +83,37 @@ void convertsOpenApiApiKeyScheme() { assertEquals("X-API-Key", apiKey.get("name")); } + @Test + @SuppressWarnings("unchecked") + void normalizedApiKeySchemeDeserializesWithItsLocation() { + Map card = + Map.of( + "securitySchemes", + Map.of("apiKeyAuth", Map.of("type", "apiKey", "in", "header", "name", "X-API-Key"))); + Map normalized = AgentCardNormalizer.normalize(card); + Map schemes = (Map) normalized.get("securitySchemes"); + SecurityScheme scheme = + new ObjectMapper() + .registerModule(new AgentCardJacksonModule()) + .convertValue(schemes.get("apiKeyAuth"), SecurityScheme.class); + + APIKeySecurityScheme apiKey = assertInstanceOf(APIKeySecurityScheme.class, scheme); + assertEquals(APIKeySecurityScheme.Location.HEADER, apiKey.location()); + assertEquals("X-API-Key", apiKey.name()); + } + + @Test + void securityRequirementScopesSurviveDeserialization() { + SecurityRequirement requirement = + new ObjectMapper() + .registerModule(new AgentCardJacksonModule()) + .convertValue( + Map.of("schemes", Map.of("oauth", List.of("tasks:read", "tasks:write"))), + SecurityRequirement.class); + + assertEquals(List.of("tasks:read", "tasks:write"), requirement.schemes().get("oauth")); + } + @Test void doesNotInventSecurityRequirementsFromSupportedSchemes() { Map card = diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCredentialServiceTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCredentialServiceTest.java index ca4ff1d8..a8399217 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCredentialServiceTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/AgentCredentialServiceTest.java @@ -5,6 +5,7 @@ package dev.openan.workflow.engine.client; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import com.sun.net.httpserver.HttpServer; @@ -13,6 +14,10 @@ import java.nio.charset.StandardCharsets; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; import org.junit.jupiter.api.Test; @@ -53,4 +58,83 @@ void formLoginEncodesKeysAndValuesAsUtf8() throws Exception { server.stop(0); } } + + @Test + void concurrentCacheMissesShareOneLogin() throws Exception { + AtomicInteger logins = new AtomicInteger(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/token", + exchange -> { + logins.incrementAndGet(); + try { + Thread.sleep(100); + } catch (InterruptedException error) { + Thread.currentThread().interrupt(); + } + byte[] response = "{\"access_token\":\"shared-token\"}".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + exchange.close(); + }); + server.start(); + var executor = Executors.newFixedThreadPool(8); + try { + Map scheme = new HashMap<>(); + scheme.put("login_url", "http://127.0.0.1:" + server.getAddress().getPort() + "/token"); + scheme.put("token_field", "access_token"); + scheme.put("request_fields", Map.of("username", "user", "password", "password")); + AgentCredentialService service = + new AgentCredentialService("agent", Map.of("bearer", scheme), HttpClient.newHttpClient()); + CountDownLatch ready = new CountDownLatch(8); + CountDownLatch start = new CountDownLatch(1); + var calls = + java.util.stream.IntStream.range(0, 8) + .mapToObj( + ignored -> + executor.submit( + () -> { + ready.countDown(); + start.await(); + return service.getCredential( + "bearer", new ClientCallContext(new HashMap<>(), new HashMap<>())); + })) + .toList(); + assertTrue(ready.await(2, TimeUnit.SECONDS)); + start.countDown(); + + for (var call : calls) assertEquals("shared-token", call.get(3, TimeUnit.SECONDS)); + assertEquals(1, logins.get()); + } finally { + executor.shutdownNow(); + server.stop(0); + } + } + + @Test + void configuredTokenFieldDoesNotSilentlyFallbackToAnotherResponseField() throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/token", + exchange -> { + byte[] response = "{\"token\":\"wrong-field\"}".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + exchange.close(); + }); + server.start(); + try { + Map scheme = new HashMap<>(); + scheme.put("login_url", "http://127.0.0.1:" + server.getAddress().getPort() + "/token"); + scheme.put("token_field", "accessSession"); + scheme.put("request_fields", Map.of("username", "user", "password", "password")); + AgentCredentialService service = + new AgentCredentialService("agent", Map.of("bearer", scheme), HttpClient.newHttpClient()); + + assertNull( + service.getCredential("bearer", new ClientCallContext(new HashMap<>(), new HashMap<>()))); + } finally { + server.stop(0); + } + } } 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 new file mode 100644 index 00000000..73304fc2 --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/CredentialHeaderContributorTest.java @@ -0,0 +1,115 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import java.net.InetSocketAddress; +import java.net.http.HttpClient; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import org.a2aproject.sdk.spec.AgentCard; +import org.junit.jupiter.api.Test; + +class CredentialHeaderContributorTest { + + private static AgentCard card(String requirements) throws Exception { + String json = + """ + { + "name":"agent","description":"test","version":"1", + "capabilities":{"streaming":false}, + "defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],"skills":[], + "securitySchemes":{ + "missing":{"httpAuthSecurityScheme":{"scheme":"Bearer"}}, + "first":{"httpAuthSecurityScheme":{"scheme":"Bearer"}}, + "second":{"httpAuthSecurityScheme":{"scheme":"Bearer"}} + }, + "securityRequirements":%s, + "supportedInterfaces":[{"protocolBinding":"HTTP+JSON","protocolVersion":"1.0","url":"https://agent.example/a2a","tenant":""}] + } + """ + .formatted(requirements); + return new ObjectMapper() + .registerModule(new AgentCardJacksonModule()) + .readValue(json, AgentCard.class); + } + + 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); + } + + @Test + void triesLaterOrRequirementWhenEarlierAlternativeIsNotConfigured() throws Exception { + HttpServer server = tokenServer(); + try { + String url = "http://127.0.0.1:" + server.getAddress().getPort() + "/token"; + AgentAuthManager manager = + new AgentAuthManager( + Map.of("agent", Map.of("second", scheme(url, "X-Second"))), + HttpClient.newHttpClient()); + CredentialHeaderContributor contributor = new CredentialHeaderContributor(manager, null); + + Map headers = + contributor.contribute( + card("[{\"schemes\":{\"missing\":{}}},{\"schemes\":{\"second\":{}}}]"), + "agent", + Map.of(), + Map.of()); + + assertEquals(Map.of("X-Second", "token"), headers); + } finally { + server.stop(0); + } + } + + @Test + void satisfiesEverySchemeInAnAndRequirement() throws Exception { + HttpServer server = tokenServer(); + try { + String url = "http://127.0.0.1:" + server.getAddress().getPort() + "/token"; + AgentAuthManager manager = + new AgentAuthManager( + Map.of( + "agent", + Map.of( + "first", scheme(url, "X-First"), + "second", scheme(url, "X-Second"))), + HttpClient.newHttpClient()); + CredentialHeaderContributor contributor = new CredentialHeaderContributor(manager, null); + + Map headers = + contributor.contribute( + card("[{\"schemes\":{\"first\":{},\"second\":{}}}]"), + "agent", + Map.of(), + Map.of()); + + assertEquals(Map.of("X-First", "token", "X-Second", "token"), headers); + } finally { + server.stop(0); + } + } + + private static HttpServer tokenServer() throws Exception { + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/token", + exchange -> { + byte[] response = "{\"access_token\":\"token\"}".getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(200, response.length); + exchange.getResponseBody().write(response); + exchange.close(); + }); + server.start(); + return server; + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/JdkHttpClientFactoryTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/JdkHttpClientFactoryTest.java new file mode 100644 index 00000000..2cdbdbd2 --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/JdkHttpClientFactoryTest.java @@ -0,0 +1,22 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.client; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.net.http.HttpClient; +import java.time.Duration; +import org.junit.jupiter.api.Test; + +class JdkHttpClientFactoryTest { + @Test + void doesNotAutomaticallyRedirectAuthenticatedProtocolRequests() { + HttpClient client = + JdkHttpClientFactory.create( + true, null, null, null, null, null, Duration.ofSeconds(1), null); + + assertEquals(HttpClient.Redirect.NEVER, client.followRedirects()); + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProtocolResponsesTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProtocolResponsesTest.java index d5277595..9fb3988f 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProtocolResponsesTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/ProtocolResponsesTest.java @@ -140,4 +140,27 @@ void metadataOnlyAndNestedDataRemainAvailable() { .get(0) .outputs()); } + + @Test + void incrementalDeliveryDoesNotReplayOrRetainPriorStandaloneMessages() { + ProtocolResponses.Accumulator accumulator = new ProtocolResponses.Accumulator(); + Message first = + Message.builder() + .messageId("first") + .role(Message.Role.ROLE_AGENT) + .parts(new TextPart("one")) + .build(); + Message second = + Message.builder() + .messageId("second") + .role(Message.Role.ROLE_AGENT) + .parts(new TextPart("two")) + .build(); + + assertEquals( + List.of("one"), accumulator.acceptIncrementally(new MessageEvent(first)).get(0).outputs()); + assertEquals( + List.of("two"), accumulator.acceptIncrementally(new MessageEvent(second)).get(0).outputs()); + assertTrue(accumulator.snapshots().isEmpty()); + } } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RuntimeChannelSelectionTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RuntimeChannelSelectionTest.java new file mode 100644 index 00000000..3272e2d6 --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/RuntimeChannelSelectionTest.java @@ -0,0 +1,30 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.client; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.HashMap; +import java.util.Map; +import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext; +import org.junit.jupiter.api.Test; + +class RuntimeChannelSelectionTest { + @Test + void channelSelectionUsesExplicitLocalStateRatherThanBusinessMetadata() { + ClientCallContext taskContext = new ClientCallContext(new HashMap<>(), new HashMap<>()); + assertFalse(DefaultA2AJavaClientRuntime.isNotificationStream(taskContext)); + + ClientCallContext notificationContext = + new ClientCallContext( + new HashMap<>( + Map.of( + A2AJavaClientRuntime.CHANNEL_STATE_KEY, + A2AJavaClientRuntime.NOTIFICATION_CHANNEL)), + new HashMap<>()); + assertTrue(DefaultA2AJavaClientRuntime.isNotificationStream(notificationContext)); + } +} 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 new file mode 100644 index 00000000..1a5429f9 --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/TransportFailuresTest.java @@ -0,0 +1,24 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * SPDX-License-Identifier: Apache-2.0 + */ +package dev.openan.workflow.engine.client; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.channels.AsynchronousCloseException; +import org.junit.jupiter.api.Test; + +class TransportFailuresTest { + @Test + void prefersExceptionTypeAndStillSupportsKnownProviderFallbacks() { + assertTrue( + TransportFailures.isExpectedLocalClose( + new RuntimeException("wrapper", new AsynchronousCloseException()))); + assertTrue( + TransportFailures.isExpectedLocalClose(new IOException("Connection closed locally"))); + assertFalse(TransportFailures.isExpectedLocalClose(new IOException("connection reset by peer"))); + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/ProtocolResultAdapterTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/ProtocolResultAdapterTest.java index b5afa91b..2611a8bf 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/ProtocolResultAdapterTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/ProtocolResultAdapterTest.java @@ -99,6 +99,24 @@ void keepsFailedProtocolStateOutOfBusinessOutputs() { assertEquals("Agent returned state=TASK_STATE_FAILED", response.getError()); } + @Test + void unknownStateEndingInUnspecifiedIsNotTreatedAsSuccess() { + var response = + ProtocolResultAdapter.toTaskResult( + SendMessageResult.builder() + .taskState("VENDOR_UNSPECIFIED") + .receivedMessages( + List.of( + new dev.openan.workflow.engine.model.ReceivedMessage( + dev.openan.workflow.engine.model.MessageContent.text("result"), + Map.of(), + List.of()))) + .build()); + + assertFalse(response.isSuccess()); + assertEquals("Agent returned state=VENDOR_UNSPECIFIED", response.getError()); + } + @Test void failedStatusIsRetainedAsEvidenceButOnlyArtifactsBecomePartialOutputs() { var status = dev.openan.workflow.engine.model.MessageContent.text("error: remote rejected"); diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java index 1dbe03f5..e1bfc4b6 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/WorkflowExecutorTest.java @@ -94,6 +94,30 @@ public CompletableFuture onRoute(RouteRequest routeRequest) { }; } + @Test + void constructorFailsFastWhenClientCallbackWiringFails() { + Workflow workflow = + Workflow.builder() + .name("wiring") + .steps( + List.of( + WorkflowStep.builder().name("s1").subtasks(List.of(task("A", "do A"))).build())) + .build(); + StubWorkflowEngineClient client = + new StubWorkflowEngineClient("A") { + @Override + public void setControlPoint(ControlPoint controlPoint) { + throw new IllegalStateException("wiring failed"); + } + }; + + IllegalStateException error = + assertThrows( + IllegalStateException.class, + () -> new WorkflowExecutor(workflow, autoCp(), client, recordingCallback(), "", "zh")); + assertEquals("wiring failed", error.getMessage()); + } + @Test void linearWorkflowTwoSteps() { WorkflowStep s1 = @@ -334,6 +358,45 @@ public CompletableFuture onRoute(RouteRequest routeRequest) { assertEquals(1, stub.getSentCount()); } + @Test + void onRouteNullDecisionFailsWithActionableMessage() { + WorkflowStep s1 = + WorkflowStep.builder() + .name("s1") + .subtasks(List.of(task("A", "do A"))) + .next(List.of(jump("s2", "cond"))) + .build(); + WorkflowStep s2 = + WorkflowStep.builder().name("s2").subtasks(List.of(task("B", "do B"))).build(); + Workflow workflow = Workflow.builder().name("null-route").steps(List.of(s1, s2)).build(); + ControlPoint controlPoint = + new ControlPoint() { + @Override + public CompletableFuture onTask(TaskRequest request) { + return CompletableFuture.completedFuture(MessageContent.text(request.getInstruction())); + } + + @Override + public CompletableFuture onRoute(RouteRequest routeRequest) { + return CompletableFuture.completedFuture(null); + } + }; + + ExecutionResult result = + new WorkflowExecutor( + workflow, + controlPoint, + new StubWorkflowEngineClient("A", "B"), + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("onRoute returned null decision")); + } + @Test void taskFailurePropagatesAndStopsWorkflow() { WorkflowStep s1 = @@ -585,6 +648,25 @@ void cyclicWorkflowFailsBeforeExecution() { assertEquals(0, stub.getSentCount()); } + @Test + void reservedTerminalRouteCannotAlsoBeAWorkflowStep() { + Workflow workflow = + Workflow.builder() + .name("reserved") + .steps( + List.of( + WorkflowStep.builder().name("end").subtasks(List.of(task("A", "run"))).build())) + .build(); + StubWorkflowEngineClient stub = new StubWorkflowEngineClient("A"); + + ExecutionResult result = + new WorkflowExecutor(workflow, autoCp(), stub, recordingCallback(), "", "zh").run().join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("reserved for a terminal route")); + assertEquals(0, stub.getSentCount()); + } + @Test void mergeDoesNotWaitForUnselectedConditionalBranch() { WorkflowStep start = diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/model/WorkflowModelTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/WorkflowModelTest.java index 54c94ddb..10bdaa6b 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/model/WorkflowModelTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/WorkflowModelTest.java @@ -118,15 +118,16 @@ void stepTypeDefaultsToAllSuccess() { } @Test - void stepTypeBogusFallsBackToAllSuccess() { + void unknownStepTypeFailsClearly() { Map data = Map.of( "name", "t", "steps", List.of(Map.of("name", "s", "step_type", "bogus", "subtasks", List.of()))); - Workflow wf = Workflow.fromMap(data); - assertEquals(StepType.ALL_SUCCESS, wf.getSteps().get(0).getStepType()); + IllegalArgumentException error = + assertThrows(IllegalArgumentException.class, () -> Workflow.fromMap(data)); + assertTrue(error.getMessage().contains("Unknown workflow step type")); } @Test @@ -144,16 +145,32 @@ void contextFromSingleStringNormalizedToList() { @Test void contextFromListPreserved() { - Map data = - Map.of( - "name", - "t", - "steps", - List.of(Map.of("name", "s", "context_from", List.of("a", "b"), "subtasks", List.of()))); + java.util.ArrayList sources = new java.util.ArrayList<>(List.of("a", "b")); + Map step = new java.util.HashMap<>(); + step.put("name", "s"); + step.put("context_from", sources); + step.put("subtasks", List.of()); + Map data = Map.of("name", "t", "steps", List.of(step)); Workflow wf = Workflow.fromMap(data); + sources.add("later"); assertEquals(List.of("a", "b"), wf.getSteps().get(0).getContextFrom()); } + @Test + void invalidWorkflowFieldTypesFailClearly() { + IllegalArgumentException nameError = + assertThrows( + IllegalArgumentException.class, + () -> Workflow.fromMap(Map.of("name", 42, "steps", List.of()))); + assertEquals("name must be a string", nameError.getMessage()); + + IllegalArgumentException stepsError = + assertThrows( + IllegalArgumentException.class, + () -> Workflow.fromMap(Map.of("name", "t", "steps", List.of("not-an-object")))); + assertEquals("steps entries must be objects", stepsError.getMessage()); + } + @Test void contextFromStarPreserved() { Map data = 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 e83c0ac3..2f443f55 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 @@ -127,7 +127,11 @@ void searchThrowsOnServerError() { server.createContext( "/api/v1/orchestrate/search", exchange -> { - exchange.sendResponseHeaders(500, 0); + byte[] response = + "{\"detail\":\"temporarily unavailable\",\"accessToken\":\"private\"}" + .getBytes(java.nio.charset.StandardCharsets.UTF_8); + exchange.sendResponseHeaders(500, response.length); + exchange.getResponseBody().write(response); exchange.close(); }); server.start(); @@ -135,7 +139,12 @@ void searchThrowsOnServerError() { String baseUrl = "http://127.0.0.1:" + port; try { - assertThrows(RuntimeException.class, () -> LoadPsop.search(baseUrl, "test", 5, null, false)); + RuntimeException error = + assertThrows( + RuntimeException.class, () -> LoadPsop.search(baseUrl, "test", 5, null, false)); + assertTrue(error.getMessage().contains("temporarily unavailable")); + assertTrue(error.getMessage().contains("\"accessToken\":\"***\"")); + assertFalse(error.getMessage().contains("private")); } finally { server.stop(0); } 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 0598f511..f8efa970 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 @@ -5,8 +5,10 @@ package dev.openan.workflow.engine.registry; import static org.junit.jupiter.api.Assertions.*; + import java.net.InetSocketAddress; import java.time.Duration; +import java.util.Map; import org.junit.jupiter.api.Test; class RegistryClientTest { @@ -37,4 +39,32 @@ void deadlineIncludesStalledResponseBody() throws Exception { server.stop(0); } } + + @Test + void registrationFailureIsNotReportedAsAResult() throws Exception { + var server = com.sun.net.httpserver.HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext( + "/rest/v1/registry-center/agent-cards", + exchange -> { + byte[] response = + "{\"detail\":\"invalid card\"}".getBytes(java.nio.charset.StandardCharsets.UTF_8); + exchange.sendResponseHeaders(400, response.length); + exchange.getResponseBody().write(response); + exchange.close(); + }); + server.start(); + try { + var client = + new RegistryClient( + "http://127.0.0.1:" + server.getAddress().getPort(), true, Duration.ofSeconds(2)); + + RuntimeException error = + assertThrows( + RuntimeException.class, () -> client.registerAgentCard(Map.of("name", "invalid"))); + assertEquals( + "Registry registration returned 400: {\"detail\":\"invalid card\"}", error.getMessage()); + } finally { + server.stop(0); + } + } }