Zhoujie628 · Sep 23, 2026 · Sep 23, 2026
@@ -1,6 +1,6 @@
 # Business callback integration contract
 
-Release `0.0.9` API; A2A-T SDK `1.1.0`, A2A Java `1.2.0.Final`. The engine owns DAG scheduling, standard A2A envelopes,
+Release `0.1.0` API; A2A-T SDK `1.1.1`, A2A Java `1.2.0.Final`. The engine owns DAG scheduling, standard A2A envelopes,
 auth, transport and task waiting. The host owns content, schemas, templates, semantic validation and any LLM calls.
 The business contracts are independent of the transport runtime.
 
@@ -247,7 +247,7 @@ independent extension lifecycles, failure propagation, cancellation, and protoco
 fixtures and do not shadow SDK schemas/templates. They are release regression evidence, not certification of production
 endpoints or live model semantics. See [Integration guide](INTEGRATION_GUIDE.md) for logging and authentication.
 
-## 9. Business-side A2A-T 1.1.0 reference
+## 9. Business-side A2A-T 1.1.1 reference
 
 Content generation, validation, filling and domain decisions belong to host-agent and dispatched-agent business code,
 not the engine. Each role calls only the APIs needed for the message it owns.

@@ -11,11 +11,11 @@ Add to your `pom.xml`:
 <dependency>
     <groupId>net.openan.workflow.sdk</groupId>
     <artifactId>workflow-engine</artifactId>
-    <version>0.0.9</version>
+    <version>0.1.0</version>
 </dependency>
 ```
 
-Version `0.0.9` is published to Maven Central and contains the APIs documented in this guide.
+Version `0.1.0` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`.
 
 The engine pulls in A2A protocol transports and a2a-t-core only. Host agents generating A2A-T content explicitly add
 a2a-t-client; dispatched-agent services that validate received content also add a2a-t-server.

@@ -22,11 +22,11 @@ callbacks own A2A-T generation, semantic validation, schemas and any LLM calls.
 <dependency>
     <groupId>net.openan.workflow.sdk</groupId>
     <artifactId>workflow-engine</artifactId>
-<version>0.0.9</version>
+    <version>0.1.0</version>
 </dependency>
 ```
 
-Version `0.0.9` is published to Maven Central and contains the APIs documented in this guide.
+Version `0.1.0` is published to Maven Central. The repository currently builds `0.1.0-SNAPSHOT`.
 
 ## 4. Quick Start
 
@@ -227,18 +227,18 @@ configuration without modifying system properties. Never log plaintext or the ke
 
 **Encrypt a password**
 
-Build the jar with `mvn -pl workflow-engine -am "-Drevision=0.0.9" package`; commands below run from the repository root.
+Build the jar with `mvn -pl workflow-engine -am "-Drevision=0.1.0" package`; commands below run from the repository root.
 `set` is Windows cmd syntax (PowerShell: `$env:A2AT_CRED_KEY='...'`). This CLI needs only the SDK jar and JDK.
 Use disposable example values here: command-line passwords/keys can appear in shell history and process listings.
 For production, obtain secrets securely in the host and use the Java encryption API.
 
 ```bash
 # Option 1: set env var first
 set A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
-java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123"
+java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123"
 
 # Option 2: pass key as second argument
-java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
+java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
 ```
 
 Output:
@@ -254,7 +254,7 @@ Paste the output into the `value` field of the credentials JSON.
 1. Generate a new key: `openssl rand -hex 32`
 2. Update the host secret store / explicit `credentialEncryptionKey`, or its OS/JVM `A2AT_CRED_KEY`
 3. Re-encrypt all passwords:
-   `java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "plaintext" new-key`
+   `java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "plaintext" new-key`
 4. Update the `enc:...` results in the credentials JSON file
 
 > The `.env` file should not be committed to version control. Add it to `.gitignore`.

@@ -1,6 +1,6 @@
 # 业务回调集成契约
 
-`0.0.9` API，A2A-T SDK `1.1.0`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和
+`0.1.0` API，A2A-T SDK `1.1.1`，A2A Java `1.2.0.Final`。 引擎负责 DAG、标准 A2A 信封、认证、传输和任务等待；宿主负责消息内容、schema、模板、语义校验和
 LLM。业务契约独立于具体传输实现。
 
 ## 1. 回调接口
@@ -209,7 +209,7 @@ acknowledgement() 代表真实 ACK 或失败；超时不伪造成功。 close()
 
 执行 `mvn -B clean verify`。Reactor 覆盖回调契约、完整响应组装、协商关联、独立扩展生命周期、失败传播、取消和协议日志脱敏。测试使用受控测试数据，不覆盖 SDK schema 或模板；这些是发布回归证据，不等于生产端点或真实模型语义认证。日志与认证配置见 [集成指南](INTEGRATION_GUIDE.md)。
 
-## 9. 业务侧 A2A-T 1.1.0 调用参考
+## 9. 业务侧 A2A-T 1.1.1 调用参考
 
 引擎只调度最终 A2A 消息；生成、校验、填充和业务判断属于宿主智能体与被调度智能体的业务代码。每个角色只调用自己所有消息需要的 API。
 

@@ -10,11 +10,11 @@
 <dependency>
     <groupId>net.openan.workflow.sdk</groupId>
     <artifactId>workflow-engine</artifactId>
-    <version>0.0.9</version>
+    <version>0.1.0</version>
 </dependency>
 ```
 
-`0.0.9` 已发布到 Maven Central，包含本文说明的接口。
+`0.1.0` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。
 
 引擎会传递性引入 A2A 协议 SDK（`a2a-java-sdk-client`，含 REST、JSON-RPC、gRPC 传输） 和最小 A2A-T 核心（`a2a-t-core`）。生成 A2A-T 内容的宿主智能体显式依赖 a2a-t-client，校验接收内容的被调度智能体服务另引入 a2a-t-server。
 

@@ -20,11 +20,11 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T
 <dependency>
     <groupId>net.openan.workflow.sdk</groupId>
     <artifactId>workflow-engine</artifactId>
-<version>0.0.9</version>
+    <version>0.1.0</version>
 </dependency>
 ```
 
-`0.0.9` 已发布到 Maven Central，包含本文说明的接口。
+`0.1.0` 已发布到 Maven Central；当前仓库构建版本为 `0.1.0-SNAPSHOT`。
 
 ## 4. 快速上手
 
@@ -217,17 +217,17 @@ A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
 
 **加密密码**
 
-先执行 `mvn -pl workflow-engine -am "-Drevision=0.0.9" package`，以下命令在仓库根目录运行，仅需 SDK jar 和 JDK。
+先执行 `mvn -pl workflow-engine -am "-Drevision=0.1.0" package`，以下命令在仓库根目录运行，仅需 SDK jar 和 JDK。
 `set` 是 Windows cmd 语法，PowerShell 应使用 `$env:A2AT_CRED_KEY='...'`。
 以下参数仅用于演示；命令行口令/密钥可能进入终端历史和进程参数。生产应由集成方安全读取密钥并调用 Java 加密 API。
 
 ```bash
 # 方式一：先设置环境变量
 set A2AT_CRED_KEY=4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
-java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123"
+java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123"
 
 # 方式二：密钥作为第二个参数
-java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
+java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "Admin@123" 4f8a2b1c3d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b1c2d3e4f5a6b7c8d9e0f1a2b
 ```
 
 输出：
@@ -242,7 +242,7 @@ enc:uHQcTeKZMVNRM9Ga:o5vm4weRozBXBs04phrLq7j7+/yRVyDsrw==
 
 1. 生成新密钥：`openssl rand -hex 32`
 2. 更新集成方密钥存储及显式 `credentialEncryptionKey`，或 OS/JVM 的 `A2AT_CRED_KEY`；引擎不自动加载 `.env`
-3. 用新密钥重新加密所有密码：`java -cp workflow-engine/target/workflow-engine-0.0.9.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥`
+3. 用新密钥重新加密所有密码：`java -cp workflow-engine/target/workflow-engine-0.1.0.jar dev.openan.workflow.engine.client.CredentialCrypto "明文密码" 新密钥`
 4. 将新的 `enc:...` 结果更新到凭证 JSON 文件
 
 > `.env` 文件不应提交到版本库，建议加入 `.gitignore`。

@@ -257,6 +257,9 @@ public WorkflowEngineClientConfig build() {
       if (maxNegotiationExchanges <= 0) {
         throw new IllegalArgumentException("maxNegotiationExchanges must be positive");
       }
+      if (taskPollIntervalMillis < 100) {
+        throw new IllegalArgumentException("taskPollIntervalMillis must be at least 100ms");
+      }
       if (credentialsConfigPath != null && credentialsConfig != null) {
         throw new IllegalArgumentException(
             "credentialsConfigPath and credentialsConfig are mutually exclusive");

@@ -101,10 +101,9 @@ public WorkflowExecutor(
     this.contextBuilder = new ContextBuilder(this.workflow, runtimeIntent);
     this.lang = lang != null ? lang : "zh";
     log.info(
-        "[Executor] Workflow: {}, steps={}, intent={}, lang={}",
+        "[Executor] Workflow: {}, steps={}, lang={}",
         workflow.getName(),
         workflow.getSteps().size(),
-        runtimeIntent,
         lang);
   }
 
@@ -114,6 +113,10 @@ private static boolean isTerminalRoute(String stepName) {
 
   private static <T> void completeFrom(
       CompletableFuture<T> destination, CompletableFuture<T> source) {
+    destination.whenComplete(
+        (value, error) -> {
+          if (error != null && !source.isDone()) source.cancel(true);
+        });
     source.whenComplete(
         (value, error) -> {
           if (error != null) destination.completeExceptionally(error);
@@ -198,9 +201,11 @@ private CompletableFuture<Void> executeSteps(
         continue;
       }
       var step = workflow.getSteps().get(idx);
-      var preds = contextBuilder.getStepPredecessors(step.getName());
+      // A direct predecessor can still be inactive while an earlier branch is running and may
+      // activate it later. Wait for every active ancestor before deciding which join inputs exist.
+      var ancestors = contextBuilder.getAllPredecessors(step.getName());
       boolean activePredecessorsComplete =
-          preds.stream()
+          ancestors.stream()
               .filter(
                   predecessor -> {
                     Integer predecessorIndex = contextBuilder.findStepIndex(predecessor);
@@ -225,7 +230,7 @@ private CompletableFuture<Void> executeSteps(
                     idx -> {
                       WorkflowStep step = workflow.getSteps().get(idx);
                       List<String> missing =
-                          contextBuilder.getStepPredecessors(step.getName()).stream()
+                          contextBuilder.getAllPredecessors(step.getName()).stream()
                               .filter(
                                   predecessor -> {
                                     Integer predecessorIndex =
@@ -368,6 +373,10 @@ private CompletableFuture<TaskResult> dispatchTask(WorkflowStep step, TaskReques
             var prepared =
                 java.util.Objects.requireNonNull(
                     controlPoint.onTask(request), "onTask returned null future");
+            result.whenComplete(
+                (value, error) -> {
+                  if (error != null && !prepared.isDone()) prepared.cancel(true);
+                });
             prepared.whenComplete(
                 (content, error) -> {
                   if (result.isDone() || stopped.get()) {
@@ -690,7 +699,13 @@ private StepResult processTaskResult(
             "status",
             task.getStatus().getValue()));
     String status = response.isSuccess() ? "success" : "failed";
-    log.info("[Executor] Task {} -> {}: {}", task.getDescription(), task.getAgent(), status);
+    log.info(
+        "[Executor] Task executionId={}, step={}, taskId={}, agent={}, status={}",
+        executionId,
+        step.getName(),
+        taskId(step.getName(), subtaskIndex),
+        task.getAgent(),
+        status);
     if (!response.isSuccess()) {
       log.warn(
           "[Executor] TASK_FAILED executionId={}, step={}, taskId={}, agent={}, errorCode={},"
@@ -704,9 +719,6 @@ private StepResult processTaskResult(
               .replace("\r", "\\r")
               .replace("\n", "\\n"));
     }
-    if (response.isSuccess() && !response.getOutputs().isEmpty()) {
-      log.debug("[Executor] Task outputs from {}: [{}]", task.getAgent(), response.getOutputs());
-    }
     List<Object> outputs = response.getOutputs();
     executionHistory.add(
         Map.of(
@@ -893,25 +905,36 @@ private RouteEvaluation evaluateRoute(
               routeEvaluationFailure(
                   step, edge, new NullPointerException("onRoute returned null future")));
     }
+    CompletableFuture<RouteDecision> callbackDecision = decision;
+    CompletableFuture<RouteDecision> bounded = new CompletableFuture<>();
+    activeTasks.add(bounded);
+    bounded.whenComplete(
+        (value, error) -> {
+          activeTasks.remove(bounded);
+          if (error != null && !callbackDecision.isDone()) callbackDecision.cancel(true);
+        });
+    callbackDecision.whenComplete(
+        (value, error) -> {
+          if (error != null) bounded.completeExceptionally(error);
+          else bounded.complete(value);
+        });
+    bounded.orTimeout(engineClient.callbackTimeoutSeconds(), TimeUnit.SECONDS);
+    if (stopped.get()) bounded.cancel(true);
     return new RouteEvaluation(
         edge,
         true,
-        decision
-            .orTimeout(engineClient.callbackTimeoutSeconds(), TimeUnit.SECONDS)
-            .handle(
-                (value, error) -> {
-                  if (error != null) {
-                    throw new CompletionException(routeEvaluationFailure(step, edge, error));
-                  }
-                  if (value == null) {
-                    throw new CompletionException(
-                        routeEvaluationFailure(
-                            step,
-                            edge,
-                            new NullPointerException("onRoute returned null decision")));
-                  }
-                  return value;
-                }));
+        bounded.handle(
+            (value, error) -> {
+              if (error != null) {
+                throw new CompletionException(routeEvaluationFailure(step, edge, error));
+              }
+              if (value == null) {
+                throw new CompletionException(
+                    routeEvaluationFailure(
+                        step, edge, new NullPointerException("onRoute returned null decision")));
+              }
+              return value;
+            }));
   }
 
   private List<Integer> collectAllowedTargets(
@@ -936,11 +959,10 @@ private void emitRouteDecision(
     String condition =
         evaluation.edge().getCondition() == null ? "" : evaluation.edge().getCondition();
     log.info(
-        "Route edge '{} -> {}': allowed={} ({})",
+        "Route edge '{} -> {}': allowed={}",
         step.getName(),
         evaluation.edge().getStep(),
-        decision.allowed(),
-        decision.reason());
+        decision.allowed());
     emit(
         EventType.ROUTE_DECISION,
         Map.of(

@@ -107,11 +107,29 @@ private static List<Task> parseSubtasks(List<Map<String, Object>> stList) {
               .agent(stringValue(t, "agent", ""))
               .skill(stringValue(t, "skill", ""))
               .description(stringValue(t, "description", ""))
+              .input(parseBusinessInput(t.get("input")))
               .build());
     }
     return subtasks;
   }
 
+  private static BusinessInput parseBusinessInput(Object raw) {
+    if (raw == null) return null;
+    if (!(raw instanceof Map<?, ?> input)) {
+      throw new IllegalArgumentException("subtask input must be an object with text or data");
+    }
+    if (input.size() != 1 || input.containsKey("text") == input.containsKey("data")) {
+      throw new IllegalArgumentException("subtask input must contain exactly one of text or data");
+    }
+    Object value = input.get(input.containsKey("text") ? "text" : "data");
+    if (value == null) {
+      throw new IllegalArgumentException("subtask input value must not be null");
+    }
+    return input.containsKey("text")
+        ? BusinessInput.text(requireString(value, "subtask input text"))
+        : BusinessInput.data(value);
+  }
+
   private static List<JumpCondition> parseNextSteps(List<Map<String, Object>> jcList) {
     List<JumpCondition> nextList = new ArrayList<>();
     for (Map<String, Object> jc : jcList) {

@@ -116,11 +116,7 @@ public static List<WorkflowSearchResult> search(
           .append(URLEncoder.encode(accessToken, StandardCharsets.UTF_8));
     }
     String url = urlBuilder.toString();
-    log.info(
-        "[Registry] Searching PSOP at {} (intent={}, top_n={})",
-        anonymousUrl(url, accessToken),
-        intent,
-        topN);
+    log.info("[Registry] Searching PSOP at {} (top_n={})", anonymousUrl(url, accessToken), topN);
     String jsonBody = mapper.writeValueAsString(Map.of("intent", intent, "top_n", topN));
     HttpResult resp = execute("POST", url, jsonBody, sslVerify, timeouts);
     if (resp.statusCode() != 200) {

@@ -115,10 +115,7 @@ public static CompletableFuture<ExecutionResult> execute(
               runtimeIntent != null ? runtimeIntent : "",
               lang != null ? lang : "zh");
       log.info(
-          "[execute_psop] Starting: workflow={}, {} steps, intent={}",
-          psop.getName(),
-          psop.getSteps().size(),
-          runtimeIntent);
+          "[execute_psop] Starting: workflow={}, {} steps", psop.getName(), psop.getSteps().size());
       collectingCallback.onEvent(
           EventType.START, Map.of("workflow", psop.getName(), "steps", psop.getSteps().size()));
       execution = executor.run();

@@ -97,17 +97,17 @@ else if (c == '}' || c == ']') {
   public static boolean sensitive(String name) {
     String key = name.toLowerCase(Locale.ROOT);
     if (key.contains("://")) return false;
-    return key.contains("authorization")
+    String normalized = key.replace("-", "").replace("_", "");
+    return normalized.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");
+        || normalized.contains("token")
+        || normalized.contains("secret")
+        || normalized.contains("password")
+        || normalized.equals("pwd")
+        || normalized.equals("passwd")
+        || normalized.equals("accesssession")
+        || normalized.equals("xapikey")
+        || normalized.equals("apikey");
   }
 
   public static Map<String, List<String>> safeHeaders(Map<String, List<String>> headers) {