fix(engine): harden protocol runtime and workflow validation

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.

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<String, String> taskTrace(String agentName, String taskId) {
     Map<String, String> 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 {

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<String, java.util.List<String>>();
       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<String>();
+            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);

workflow-engine/src/main/java/dev/openan/workflow/engine/client/AgentCredentialService.java

@@ -61,6 +61,8 @@ class AgentCredentialService implements CredentialService {
 
   private final Map<String, TokenEntry> tokenCache = new ConcurrentHashMap<>();
 
+  private final Map<String, Object> refreshLocks = new ConcurrentHashMap<>();
+
   public AgentCredentialService(String agentName, Map<String, Map<String, Object>> 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<String, Object> schemeCfg) {
@@ -245,6 +253,10 @@ private String login(Map<String, Object> schemeCfg) {
               .build();
       HttpResponse<String> 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<Map<String, Object>>() {});
     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;
   }

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<String, String> headers,
       Map<String, Object> 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<String, String> 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<String, String> contribute(
           "Authentication failed for agent " + agentName + ": credential configuration is empty");
     }
 
+    java.util.ArrayList<String> failures = new java.util.ArrayList<>();
     for (SecurityRequirement requirement : securityRequirements) {
+      Map<String, String> candidate = new HashMap<>();
       Map<String, List<String>> 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<String, Object> 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)
+            + ")");
   }
 }

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);

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<ClientEvent> 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);