From d7dd71ea5d2fb188d7bf48b64ac2e7d424271366 Mon Sep 17 00:00:00 2001 From: hanyu0504 Date: Wed, 23 Sep 2026 15:27:58 +0800 Subject: [PATCH] fix(engine): harden workflow joins, cancellation and logging Signed-off-by: hanyu0504 --- docs/en/BUSINESS_CALLBACKS.md | 4 +- docs/en/DEVELOPER_GUIDE.md | 4 +- docs/en/INTEGRATION_GUIDE.md | 12 +- docs/zh/BUSINESS_CALLBACKS.md | 4 +- docs/zh/DEVELOPER_GUIDE.md | 4 +- docs/zh/INTEGRATION_GUIDE.md | 12 +- .../client/WorkflowEngineClientConfig.java | 3 + .../engine/core/WorkflowExecutor.java | 78 +++++++----- .../workflow/engine/model/Workflow.java | 18 +++ .../workflow/engine/registry/LoadPsop.java | 6 +- .../workflow/engine/runner/ExecutePsop.java | 5 +- .../engine/util/SensitiveDataRedactor.java | 20 +-- .../engine/client/WireLogRedactionTest.java | 10 ++ .../WorkflowEngineClientConfigTest.java | 9 ++ .../engine/core/WorkflowExecutorTest.java | 114 ++++++++++++++++++ .../engine/model/WorkflowModelTest.java | 36 ++++++ 16 files changed, 272 insertions(+), 67 deletions(-) diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index 7096e7d7..04c70132 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -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. diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index a71932e6..0094ed4c 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -11,11 +11,11 @@ Add to your `pom.xml`: net.openan.workflow.sdk workflow-engine - 0.0.9 + 0.1.0 ``` -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. diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index d183aaa8..f8d0ba57 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -22,11 +22,11 @@ callbacks own A2A-T generation, semantic validation, schemas and any LLM calls. net.openan.workflow.sdk workflow-engine -0.0.9 + 0.1.0 ``` -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,7 +227,7 @@ 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. @@ -235,10 +235,10 @@ For production, obtain secrets securely in the host and use the Java encryption ```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`. diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index 465d5348..e5fac0c7 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -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。 diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index fcbed543..88258295 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -10,11 +10,11 @@ net.openan.workflow.sdk workflow-engine - 0.0.9 + 0.1.0 ``` -`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。 diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 3080aa46..eb490ae1 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -20,11 +20,11 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T net.openan.workflow.sdk workflow-engine -0.0.9 + 0.1.0 ``` -`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`。 diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java index 9b41b7e3..d3d534b4 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfig.java @@ -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"); 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 66c4fed0..a5c8c260 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 @@ -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 void completeFrom( CompletableFuture destination, CompletableFuture 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 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 executeSteps( idx -> { WorkflowStep step = workflow.getSteps().get(idx); List missing = - contextBuilder.getStepPredecessors(step.getName()).stream() + contextBuilder.getAllPredecessors(step.getName()).stream() .filter( predecessor -> { Integer predecessorIndex = @@ -368,6 +373,10 @@ private CompletableFuture 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 outputs = response.getOutputs(); executionHistory.add( Map.of( @@ -893,25 +905,36 @@ private RouteEvaluation evaluateRoute( routeEvaluationFailure( step, edge, new NullPointerException("onRoute returned null future"))); } + CompletableFuture callbackDecision = decision; + CompletableFuture 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 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( 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 fed41b11..3017bbf4 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 @@ -107,11 +107,29 @@ private static List parseSubtasks(List> 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 parseNextSteps(List> jcList) { List nextList = new ArrayList<>(); for (Map jc : jcList) { 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 0180d9b9..5df62bd2 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 @@ -116,11 +116,7 @@ public static List search( .append(URLEncoder.encode(accessToken, StandardCharsets.UTF_8)); } String url = urlBuilder.toString(); - log.info( - "[Registry] Searching PSOP at {} (intent={}, top_n={})", - anonymousUrl(url, accessToken), - intent, - topN); + log.info("[Registry] Searching PSOP at {} (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) { diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java index 5276be8c..e196ac4b 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/runner/ExecutePsop.java @@ -115,10 +115,7 @@ public static CompletableFuture 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(); 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 index 2079368f..f200c53c 100644 --- 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 @@ -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> safeHeaders(Map> headers) { diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WireLogRedactionTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WireLogRedactionTest.java index 56025055..b3e5827c 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WireLogRedactionTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WireLogRedactionTest.java @@ -26,6 +26,16 @@ import org.junit.jupiter.api.Test; class WireLogRedactionTest { + @Test + void masksUnderscoredCredentialNamesInJsonAndHeaders() { + assertEquals( + "{\"api_key\":\"***\",\"access_session\":\"***\"}", + WireLog.redact("{\"api_key\":\"hidden\",\"access_session\":\"hidden\"}")); + assertEquals( + List.of("***"), + WireLog.safeHeaders(Map.of("X-Api_Key", List.of("hidden"))).get("X-Api_Key")); + } + @Test void masksAuthorizationValuesInRemoteErrorText() { assertEquals( diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java index 601b583a..fa05a031 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/client/WorkflowEngineClientConfigTest.java @@ -57,6 +57,15 @@ void notificationAcknowledgementDefaultsToFiveMinutesAndRemainsConfigurable() { @Test void rejectsInvalidResourceLimits() { + assertThrows( + IllegalArgumentException.class, + () -> WorkflowEngineClientConfig.builder().taskPollIntervalMillis(99).build()); + assertEquals( + 100, + WorkflowEngineClientConfig.builder() + .taskPollIntervalMillis(100) + .build() + .getTaskPollIntervalMillis()); assertThrows( IllegalArgumentException.class, () -> WorkflowEngineClientConfig.builder().maxNegotiationExchanges(0).build()); 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 8fb98f3f..54c372a5 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 @@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import dev.openan.workflow.engine.StubWorkflowEngineClient; @@ -91,6 +92,119 @@ private JumpCondition jump(String step, String cond) { return JumpCondition.builder().step(step).condition(cond).build(); } + @Test + void mergeWaitsForActiveAncestorsBeforeItsDirectPredecessorIsActivated() { + List mergeSources = new ArrayList<>(); + List steps = new ArrayList<>(); + for (String name : List.of("short", "long1", "long2", "long3", "merge")) { + String next = + switch (name) { + case "short", "long3" -> "merge"; + case "long1" -> "long2"; + case "long2" -> "long3"; + default -> null; + }; + steps.add( + WorkflowStep.builder() + .name(name) + .stepType(StepType.SELF_LOOP) + .subtasks(List.of(task("local", name))) + .next(next == null ? List.of() : List.of(jump(next, ""))) + .build()); + } + ControlPoint callbacks = + ControlPoint.builder() + .onSelfTask( + request -> { + if (request.getStepName().equals("merge")) { + request + .getWorkflowInput() + .upstreamResults() + .forEach(result -> mergeSources.add(result.stepName())); + } + return CompletableFuture.completedFuture( + TaskResult.success(List.of(request.getStepName()))); + }) + .build(); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().steps(steps).build(), + callbacks, + new StubWorkflowEngineClient(), + null, + "", + "en") + .run() + .join(); + + assertTrue(result.isSuccess()); + assertEquals(Set.of("short", "long3"), new HashSet<>(mergeSources)); + assertEquals(5, result.getHistory().size()); + } + + @Test + void cancellationPropagatesToPendingLocalCallback() throws Exception { + CompletableFuture local = new CompletableFuture<>(); + CompletableFuture entered = new CompletableFuture<>(); + ControlPoint callbacks = + ControlPoint.builder() + .onSelfTask( + request -> { + entered.complete(null); + return local; + }) + .build(); + WorkflowStep step = + WorkflowStep.builder() + .name("local") + .stepType(StepType.SELF_LOOP) + .subtasks(List.of(task("local", "work"))) + .build(); + var run = + new WorkflowExecutor( + Workflow.builder().steps(List.of(step)).build(), + callbacks, + new StubWorkflowEngineClient(), + null, + "", + "en") + .run(); + entered.get(3, TimeUnit.SECONDS); + run.cancel(true); + // The callback can return concurrently with cancellation; await propagation without sleeps. + assertThrows( + java.util.concurrent.CancellationException.class, () -> local.get(3, TimeUnit.SECONDS)); + } + + @Test + void cancellationPropagatesToPendingRouteCallback() throws Exception { + CompletableFuture route = new CompletableFuture<>(); + CompletableFuture entered = new CompletableFuture<>(); + ControlPoint callbacks = + ControlPoint.builder() + .onRoute( + request -> { + entered.complete(null); + return route; + }) + .build(); + WorkflowStep start = + WorkflowStep.builder().name("start").next(List.of(jump("end", "condition"))).build(); + var run = + new WorkflowExecutor( + Workflow.builder().steps(List.of(start)).build(), + callbacks, + new StubWorkflowEngineClient(), + null, + "", + "en") + .run(); + entered.get(3, TimeUnit.SECONDS); + run.cancel(true); + assertTrue(route.isCancelled()); + } + /** ControlPoint that prepares task content and allows every conditional edge. */ private ControlPoint autoCp() { return new ControlPoint() { 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 10bdaa6b..931ae976 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 @@ -28,6 +28,42 @@ /** Tests for Workflow/WorkflowStep/Task/JumpCondition/StepType model parsing. */ class WorkflowModelTest { + @Test + void mapLoadingPreservesStructuredBusinessInput() { + Map input = Map.of("data", Map.of("serviceId", "service-42")); + Workflow workflow = + Workflow.fromMap( + Map.of( + "steps", + List.of( + Map.of( + "name", + "start", + "subtasks", + List.of( + Map.of("agent", "A", "description", "diagnose", "input", input)))))); + assertEquals( + BusinessInput.data(Map.of("serviceId", "service-42")), + workflow.getSteps().get(0).getSubtasks().get(0).getInput()); + } + + @Test + void mapLoadingRejectsAmbiguousBusinessInput() { + Map input = Map.of("text", "diagnose", "data", Map.of("id", 42)); + assertThrows( + IllegalArgumentException.class, + () -> + Workflow.fromMap( + Map.of( + "steps", + List.of( + Map.of( + "name", + "start", + "subtasks", + List.of(Map.of("agent", "A", "input", input))))))); + } + @Test void parsesLinearWorkflow() { Map data =