diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ba786d7..a510d6d9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ ## [Unreleased] +### Breaking: per-edge conditional routing + +- `onRoute` now evaluates one conditional edge at a time: `RouteRequest` describes a single edge (step, condition, + upstream window) and the callback returns `RouteDecision.allow()` / + `deny()` for that edge. Unconditional edges always run and bypass the callback; allowed conditional edges activate in + parallel with them. This replaces the previous N-choose-1 candidate selection (`RouteDecision.nextStep` and + `RouteRequest.candidates` are removed) + and fixes mixed-edge steps, where all unconditional edges were not previously guaranteed to activate. +- Workflow validation now rejects duplicate outgoing targets on one step, null edges and blank targets. +- After all route decisions succeed, `route_decision` is emitted once per outgoing edge, including unconditional edges, + instead of once per step. A callback failure emits the workflow error and no partial per-edge decisions. + ## [0.0.4] — 2026-09-08 - Make the Spring Boot A2A server auto-configuration opt-in: server beans are registered only diff --git a/docs/en/API_REFERENCE.md b/docs/en/API_REFERENCE.md index 0992135f..cc9c9003 100644 --- a/docs/en/API_REFERENCE.md +++ b/docs/en/API_REFERENCE.md @@ -274,8 +274,9 @@ interface ControlPoint { ``` onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content. -onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop. -Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent. +onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation +returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No +echo-success, implicit route approval or automatic consent. `DefaultControlPoint` preserves those fail-fast defaults and can delegate `onNegotiation` to an injected `NegotiationStrategy`. Implement `NegotiationStrategy.resolve(NegotiationRequest)` when negotiation policy is the only @@ -295,30 +296,30 @@ Override to receive real-time execution events. Event types are defined in `Even ### EventType -| Constant | Description | -|--------------------------|--------------------------------------------------------------| -| `STEP_START` | A workflow step began | -| `STEP_COMPLETE` | A workflow step completed | -| `TASK_REQUEST` | A task was dispatched to an agent | -| `TASK_RESPONSE` | A task response was received | -| `TASK_STATUS_CHANGED` | A task's status changed (pending → running → success/failed) | -| `AGENT_REQUEST` | A message was sent to an agent | -| `AGENT_RESPONSE` | A response was received from an agent | -| `AGENT_STATUS_UPDATE` | Agent SSE status update (SUBMITTED, WORKING, etc.) | -| `AGENT_ARTIFACT_UPDATE` | Agent SSE artifact update | -| `AGENT_MESSAGE_EVENT` | Agent SSE message event | -| `NEGOTIATION_REQUEST` | Agent requested negotiation (INPUT_REQUIRED) | -| `NEGOTIATION_RESOLVED` | Clarification was sent to agent | -| `NEGOTIATION_FAILED` | Negotiation could not be resolved | -| `AUTHORIZATION_REQUEST` | Reserved; not currently emitted (authorization results arrive via the `ExtensionSender` future) | -| `AUTHORIZATION_RESOLVED` | Reserved; not currently emitted | +| Constant | Description | +|--------------------------|----------------------------------------------------------------------------------------------------------| +| `STEP_START` | A workflow step began | +| `STEP_COMPLETE` | A workflow step completed | +| `TASK_REQUEST` | A task was dispatched to an agent | +| `TASK_RESPONSE` | A task response was received | +| `TASK_STATUS_CHANGED` | A task's status changed (pending → running → success/failed) | +| `AGENT_REQUEST` | A message was sent to an agent | +| `AGENT_RESPONSE` | A response was received from an agent | +| `AGENT_STATUS_UPDATE` | Agent SSE status update (SUBMITTED, WORKING, etc.) | +| `AGENT_ARTIFACT_UPDATE` | Agent SSE artifact update | +| `AGENT_MESSAGE_EVENT` | Agent SSE message event | +| `NEGOTIATION_REQUEST` | Agent requested negotiation (INPUT_REQUIRED) | +| `NEGOTIATION_RESOLVED` | Clarification was sent to agent | +| `NEGOTIATION_FAILED` | Negotiation could not be resolved | +| `AUTHORIZATION_REQUEST` | Reserved; not currently emitted (authorization results arrive via the `ExtensionSender` future) | +| `AUTHORIZATION_RESOLVED` | Reserved; not currently emitted | | `NOTIFICATION` | Reserved; not currently emitted (subscription events arrive via the `NotificationSubscription` callback) | -| `ROUTE_DECISION` | Route decision was made | -| `WORKFLOW_COMPLETE` | DAG scheduling ended; check success, not all nodes necessarily ran | -| `START` | Workflow execution started | -| `COMPLETE` | Workflow execution completed successfully | -| `ERROR` | Workflow execution failed | -| `CLOSE` | Engine client closed | +| `ROUTE_DECISION` | Successful routing result for one outgoing edge | +| `WORKFLOW_COMPLETE` | DAG scheduling ended; check success, not all nodes necessarily ran | +| `START` | Workflow execution started | +| `COMPLETE` | Workflow execution completed successfully | +| `ERROR` | Workflow execution failed | +| `CLOSE` | Engine client closed | --- @@ -542,12 +543,26 @@ flattened metadata are transport diagnostics, not substitutes for the complete b | `stepOutputs` | `Map` | Outputs keyed by step name | | `error` | `String` | Error message (if failed) | -### RouteDecision +### RouteRequest / RouteDecision -| Field | Type | Description | -|------------|----------|----------------------| -| `nextStep` | `String` | Next step to execute | -| `reason` | `String` | Decision reason | +`RouteRequest` describes one conditional edge, not the complete outgoing-edge list: + +| Field | Type | Description | +|------------------|-----------------------------|------------------------------------------| +| `executionId` | `String` | Current workflow execution | +| `stepName` | `String` | Source step | +| `nextStep` | `String` | Target of this edge | +| `condition` | `String` | Nonblank business condition to evaluate | +| `workflowInput` | `WorkflowInput` | Selected protocol-neutral context window | +| `currentResults` | `List` | Results produced by the source step | + +| `RouteDecision` field | Type | Description | +|-----------------------|-----------|-------------------------------------| +| `allowed` | `boolean` | Whether this edge must be activated | +| `reason` | `String` | Optional business decision reason | + +Use `RouteDecision.allow(...)` or `RouteDecision.deny(...)`. Every blank-condition edge is activated without invoking +the callback; every allowed conditional edge is activated as well, so one source step can fan out to multiple targets. ### WorkflowSearchResult diff --git a/docs/en/BUSINESS_CALLBACKS.md b/docs/en/BUSINESS_CALLBACKS.md index e0fd8be9..f0869d6c 100644 --- a/docs/en/BUSINESS_CALLBACKS.md +++ b/docs/en/BUSINESS_CALLBACKS.md @@ -16,8 +16,9 @@ interface ControlPoint { ``` onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content. -onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop. -Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent. +onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation +returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No +echo-success, implicit route approval or automatic consent. See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples. ```java @@ -27,8 +28,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); @@ -145,11 +146,16 @@ SDK-specific content exception classes. ## 6. Routing, concurrency and failures -RouteRequest(executionId, stepName, workflowInput, currentResults, candidates). Candidates are RouteOption(nextStep, -condition); return RouteDecision.builder().nextStep(allowed).build(). Unconditional edges fan out without calling -onRoute; conditional selection must be an allowed destination. +RouteRequest (executionId, stepName, nextStep, condition, workflowInput, currentResults) describes one conditional edge. +Return RouteDecision.allow (reason) to activate it or RouteDecision.deny (reason) to skip it. Null, empty and +whitespace-only conditions are unconditional: they bypass onRoute and always activate. Each nonblank edge is evaluated +independently, so the result can activate zero, one or many conditional targets together with all unconditional targets. +The engine waits for every route decision before dispatching any successor; a missing, null, exceptional or timed-out +decision fails the workflow and cannot cause partial successor dispatch. Route failure messages identify the source and +target edge while preserving the original cause. -Callbacks may run concurrently; do not hold shared mutable current-task state. Each workflow task activation +Callbacks for different conditional edges may run concurrently; do not hold shared mutable current-task state. Each +workflow task activation (preparation and dispatch combined) is bounded by the client timeout (default sendTimeoutSeconds=600). Routing has a callback timeout; dispatch/negotiation additionally has its own total wait deadline. Cancellation/timeout prevents late sends; it does not automatically stop external business/LLM work. Hosts own cancellation of their resources. Synchronous diff --git a/docs/en/DESIGN.md b/docs/en/DESIGN.md index ab099dfe..5c94b729 100644 --- a/docs/en/DESIGN.md +++ b/docs/en/DESIGN.md @@ -248,8 +248,9 @@ interface ControlPoint { ``` onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content. -onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop. -Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent. +onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation +returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No +echo-success, implicit route approval or automatic consent. See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples. ## 5. A2A-T Extension Model @@ -282,12 +283,16 @@ open until the host closes them. A step's `next` list holds `JumpCondition(step, condition)` entries. The routing rule is: - **No `next`** - terminal; the step completes the branch. -- **All conditions empty** - unconditional fan-out: dispatch every non-terminal next step in parallel. -- **Has conditions** - conditional: call `ControlPoint.onRoute`, which returns a single `RouteDecision.nextStep`. The - engine enforces that the returned step is among the declared conditions; an invalid return fails the workflow with an - error. - -This makes conditional branches an N-choose-1 selection and keeps unconditional fan-out as automatic parallel dispatch. +- **Blank condition** - unconditional edge: activate it without calling `ControlPoint.onRoute`. Null, empty and + whitespace-only values are all blank. +- **Nonblank condition** - conditional edge: invoke `ControlPoint.onRoute` once for that edge and activate it only when + the returned `RouteDecision.allowed` is true. + +All conditional edges are evaluated independently and may run concurrently. The engine waits for every decision before +atomically activating all unconditional and allowed conditional targets. Multiple true decisions therefore create a +parallel fan-out; all false decisions are valid and end the branch when no unconditional target exists. Callback +failure, timeout or null fails routing before any successor is dispatched. Route failures identify the source and target +edge and preserve the original cause. Duplicate targets from the same source are rejected during graph validation. --- diff --git a/docs/en/DEVELOPER_GUIDE.md b/docs/en/DEVELOPER_GUIDE.md index 43bf3d7f..40f1c09d 100644 --- a/docs/en/DEVELOPER_GUIDE.md +++ b/docs/en/DEVELOPER_GUIDE.md @@ -38,8 +38,9 @@ interface ControlPoint { ``` onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content. -onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop. -Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent. +onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation +returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No +echo-success, implicit route approval or automatic consent. See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples. ```java @@ -49,8 +50,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); @@ -85,26 +86,26 @@ and a sync `BiConsumer` overload. Events come from three layers: the runner (lifecycle bracket), the executor (step/task/routing), and the engine client (agent traffic, negotiation). -| Event | Layer | When | Key Data | -|-------------------------|--------------------|----------------------------------------------------------|---------------------------------------------------------| -| `start` | runner | Workflow begins | `workflow`, `steps` | -| `step_start` | executor | Step begins | `step` | -| `task_request` | executor | A subtask is dispatched to `onTask`/`onSelfTask` | `step`, `agent`, `task` | -| `task_response` | executor | Remote task completed or onSelfTask returned TaskResult | `step`, `agent`, `task`, `outputs` | -| `task_status_changed` | executor | Task status changed (pending → running → success/failed) | `step`, `agent`, `task`, `status` | -| `route_decision` | executor | Branch chosen | `step`, `next`, `reason` | -| `step_complete` | executor | Step finished | `step`, `results` | -| `workflow_complete` | executor | All steps finished | `history`, `step_outputs` | -| `agent_request` | engine client | Dispatch intent, not a wire observation | `agent`, `content` | -| `agent_response` | engine client | Remote response assembled | `agent`, `response`, `receivedMessages` | -| `agent_status_update` | engine client | Agent SSE status update | `agent`, `state`, `is_final` | -| `agent_artifact_update` | engine client | Agent SSE artifact update | `agent`, `artifact_name`, `text` | -| `negotiation_request` | engine client | Valid Propose enters host callback | `agent`, `request`, `exchange` | -| `negotiation_resolved` | engine client | Host Send passed association checks, not task success | `agent`, `reply`, `exchange` | -| `negotiation_failed` | engine client | Local negotiation interaction failed | `agent`, `exchange`, `errorType` | -| `complete` | runner | Workflow succeeded | `history`, `step_outputs` | -| `error` | runner or executor | Workflow failed | runner: `error`, `history`; executor: `step`, `results` | -| `close` | runner | Cleanup done | (empty) | +| Event | Layer | When | Key Data | +|-------------------------|--------------------|----------------------------------------------------------|-----------------------------------------------------------------| +| `start` | runner | Workflow begins | `workflow`, `steps` | +| `step_start` | executor | Step begins | `step` | +| `task_request` | executor | A subtask is dispatched to `onTask`/`onSelfTask` | `step`, `agent`, `task` | +| `task_response` | executor | Remote task completed or onSelfTask returned TaskResult | `step`, `agent`, `task`, `outputs` | +| `task_status_changed` | executor | Task status changed (pending → running → success/failed) | `step`, `agent`, `task`, `status` | +| `route_decision` | executor | Successful routing recorded for one outgoing edge | `step`, `next`, `condition`, `conditional`, `allowed`, `reason` | +| `step_complete` | executor | Step finished | `step`, `results` | +| `workflow_complete` | executor | All steps finished | `history`, `step_outputs` | +| `agent_request` | engine client | Dispatch intent, not a wire observation | `agent`, `content` | +| `agent_response` | engine client | Remote response assembled | `agent`, `response`, `receivedMessages` | +| `agent_status_update` | engine client | Agent SSE status update | `agent`, `state`, `is_final` | +| `agent_artifact_update` | engine client | Agent SSE artifact update | `agent`, `artifact_name`, `text` | +| `negotiation_request` | engine client | Valid Propose enters host callback | `agent`, `request`, `exchange` | +| `negotiation_resolved` | engine client | Host Send passed association checks, not task success | `agent`, `reply`, `exchange` | +| `negotiation_failed` | engine client | Local negotiation interaction failed | `agent`, `exchange`, `errorType` | +| `complete` | runner | Workflow succeeded | `history`, `step_outputs` | +| `error` | runner or executor | Workflow failed | runner: `error`, `history`; executor: `step`, `results` | +| `close` | runner | Cleanup done | (empty) | ## 6. Mid-Level (Layer 1: WorkflowExecutor) diff --git a/docs/en/INTEGRATION_GUIDE.md b/docs/en/INTEGRATION_GUIDE.md index 5aadbff8..4378150a 100644 --- a/docs/en/INTEGRATION_GUIDE.md +++ b/docs/en/INTEGRATION_GUIDE.md @@ -34,7 +34,7 @@ Complete runnable source: [HostQuickStart.java](../../samples/src/main/java/dev/ It is compiled and its remote-task/local-aggregation flow is tested by HostQuickStartTest. Copy that source into your host project or run it from samples in IDEA (registry URL, target agent name, credentials path). The snippets below explain the same API; use one AgentCard loading option, and implement domain-specific content for Task-T. -Nonempty business conditions require an onRoute policy; this minimal workflow uses unconditional edges. +Each nonblank business condition requires an onRoute decision; this minimal workflow uses unconditional edges. ### 4.1 Define a Workflow @@ -112,8 +112,9 @@ interface ControlPoint { ``` onTask returns final parts/metadata/extensions; the engine sends them without generating or rewriting content. -onSelfTask returns local TaskResult, onRoute selects an allowed candidate, and onNegotiation returns Send or Stop. -Unimplemented callbacks fail explicitly. No echo-success, first-branch choice or automatic consent. +onSelfTask returns local TaskResult, onRoute independently allows or denies one conditional edge, and onNegotiation +returns Send or Stop. Unconditional edges bypass onRoute and always run. Unimplemented callbacks fail explicitly. No +echo-success, implicit route approval or automatic consent. See [Business callback contract](BUSINESS_CALLBACKS.md) for fields and working examples. ```java @@ -123,8 +124,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); diff --git a/docs/zh/API_REFERENCE.md b/docs/zh/API_REFERENCE.md index a7839e77..b11d20fc 100644 --- a/docs/zh/API_REFERENCE.md +++ b/docs/zh/API_REFERENCE.md @@ -261,8 +261,9 @@ interface ControlPoint { } ``` -onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。 onSelfTask 返回本地 TaskResult；onRoute -选择允许的候选；onNegotiation 返回 Send 或 Stop。 未实现的回调明确失败，不回显成功、不选首分支、不自动同意。 +onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute +对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， +不回显成功、不默认放行条件边、不自动同意。 `DefaultControlPoint` 保留上述快速失败默认值，并可把 `onNegotiation` 委托给注入的 `NegotiationStrategy`。仅需定制协商策略时实现 `NegotiationStrategy.resolve(NegotiationRequest)`；同时需要定制任务、宿主本地任务或路由时，实现或构建完整 `ControlPoint`。 @@ -281,30 +282,30 @@ public class EventCallback { ### EventType -| 常量 | 说明 | -|--------------------------|----------------------------------------------------| -| `STEP_START` | 工作流步骤开始 | -| `STEP_COMPLETE` | 工作流步骤完成 | -| `TASK_REQUEST` | 任务分派给智能体 | -| `TASK_RESPONSE` | 收到任务响应 | -| `TASK_STATUS_CHANGED` | 任务状态变更（pending → running → success/failed） | -| `AGENT_REQUEST` | 消息发送给智能体 | -| `AGENT_RESPONSE` | 收到智能体响应 | -| `AGENT_STATUS_UPDATE` | 智能体 SSE 状态更新 | -| `AGENT_ARTIFACT_UPDATE` | 智能体 SSE artifact 更新 | -| `AGENT_MESSAGE_EVENT` | 智能体 SSE 消息事件 | -| `NEGOTIATION_REQUEST` | 智能体请求协商 | -| `NEGOTIATION_RESOLVED` | 补充信息已发送 | -| `NEGOTIATION_FAILED` | 协商无法解决 | -| `AUTHORIZATION_REQUEST` | 保留；当前不上报（授权结果经 `ExtensionSender` 返回） | -| `AUTHORIZATION_RESOLVED` | 保留；当前不上报 | +| 常量 | 说明 | +|--------------------------|--------------------------------------------------------------------| +| `STEP_START` | 工作流步骤开始 | +| `STEP_COMPLETE` | 工作流步骤完成 | +| `TASK_REQUEST` | 任务分派给智能体 | +| `TASK_RESPONSE` | 收到任务响应 | +| `TASK_STATUS_CHANGED` | 任务状态变更（pending → running → success/failed） | +| `AGENT_REQUEST` | 消息发送给智能体 | +| `AGENT_RESPONSE` | 收到智能体响应 | +| `AGENT_STATUS_UPDATE` | 智能体 SSE 状态更新 | +| `AGENT_ARTIFACT_UPDATE` | 智能体 SSE artifact 更新 | +| `AGENT_MESSAGE_EVENT` | 智能体 SSE 消息事件 | +| `NEGOTIATION_REQUEST` | 智能体请求协商 | +| `NEGOTIATION_RESOLVED` | 补充信息已发送 | +| `NEGOTIATION_FAILED` | 协商无法解决 | +| `AUTHORIZATION_REQUEST` | 保留；当前不上报（授权结果经 `ExtensionSender` 返回） | +| `AUTHORIZATION_RESOLVED` | 保留；当前不上报 | | `NOTIFICATION` | 保留；当前不上报（订阅事件经 `NotificationSubscription` 回调送达） | -| `ROUTE_DECISION` | 路由决策已做出 | -| `WORKFLOW_COMPLETE` | DAG 调度结束，需检查 success；不代表全部节点成功执行 | -| `START` | 工作流执行开始 | -| `COMPLETE` | 工作流执行成功完成 | -| `ERROR` | 工作流执行失败 | -| `CLOSE` | 引擎客户端已关闭 | +| `ROUTE_DECISION` | 路由阶段成功后记录一条出边的放行或拒绝结果 | +| `WORKFLOW_COMPLETE` | DAG 调度结束，需检查 success；不代表全部节点成功执行 | +| `START` | 工作流执行开始 | +| `COMPLETE` | 工作流执行成功完成 | +| `ERROR` | 工作流执行失败 | +| `CLOSE` | 引擎客户端已关闭 | --- @@ -514,12 +515,26 @@ getReceivedMessages() 是保留层级的响应来源，getOutputs() 为便利投 | `stepOutputs` | `Map` | 按步骤名索引的输出 | | `error` | `String` | 错误信息（失败时） | -### RouteDecision +### RouteRequest / RouteDecision -| 字段 | 类型 | 说明 | -|------------|----------|------------| -| `nextStep` | `String` | 下一步执行 | -| `reason` | `String` | 决策原因 | +`RouteRequest` 每次描述一条条件边，不再携带完整出边候选列表： + +| 字段 | 类型 | 说明 | +|------------------|-----------------------------|----------------------| +| `executionId` | `String` | 当前工作流执行标识 | +| `stepName` | `String` | 源节点 | +| `nextStep` | `String` | 当前边的目标节点 | +| `condition` | `String` | 待判断的非空业务条件 | +| `workflowInput` | `WorkflowInput` | 协议无关的上下文窗口 | +| `currentResults` | `List` | 源节点产生的任务结果 | + +| `RouteDecision` 字段 | 类型 | 说明 | +|----------------------|-----------|--------------------| +| `allowed` | `boolean` | 是否激活当前条件边 | +| `reason` | `String` | 可选的业务判断原因 | + +通过 `RouteDecision.allow(...)` 或 `RouteDecision.deny(...)` 返回判断。所有空条件边直接激活，所有返回 allow 的条件边也会激活， +所以一个源节点可以同时扇出到多个目标节点。 ### WorkflowSearchResult diff --git a/docs/zh/BUSINESS_CALLBACKS.md b/docs/zh/BUSINESS_CALLBACKS.md index ebcb6439..0dae4450 100644 --- a/docs/zh/BUSINESS_CALLBACKS.md +++ b/docs/zh/BUSINESS_CALLBACKS.md @@ -14,8 +14,9 @@ interface ControlPoint { } ``` -onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。 onSelfTask 返回本地 TaskResult；onRoute -选择允许的候选；onNegotiation 返回 Send 或 Stop。 未实现的回调明确失败，不回显成功、不选首分支、不自动同意。 +onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute +对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， +不回显成功、不默认放行条件边、不自动同意。 字段与完整示例见 [业务回调集成契约](BUSINESS_CALLBACKS.md)。 ```java @@ -25,8 +26,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); @@ -125,10 +126,13 @@ safeMessage, safeDetails)； 引擎不识别 SDK 专属内容异常类，也不 ## 6. 路由、并发与失败 -RouteRequest(executionId, stepName, workflowInput, currentResults, candidates)。 候选为 RouteOption(nextStep, -condition)，返回 RouteDecision.builder().nextStep(允许目标).build()。 无条件边自动并行推进，条件分支只接受候选中的目标。 +RouteRequest (executionId, stepName, nextStep, condition, workflowInput, currentResults) 每次描述一条条件边。 返回 +RouteDecision.allow (reason) 激活该边，或返回 RouteDecision.deny (reason) 跳过该边。null、空字符串和纯空白条件都是无条件边， +不进入 onRoute 并始终激活。每条非空条件边独立判断，因此最终可同时激活零条、一条或多条条件边，并与所有无条件边一起推进。 +引擎等待全部路由判断完成后才下发任何后继节点；回调缺失、返回 null、异常或超时都会使工作流失败，不会发生部分后继任务已下发。 +路由失败信息会标明源节点和目标节点，并保留原始异常原因。 -回调可能并发，不要共享可变的“当前任务”状态。每个工作流任务从内容准备到传输完成受客户端 timeout 总体限制，默认 +不同条件边的回调可能并发，不要共享可变的“当前任务”状态。每个工作流任务从内容准备到传输完成受客户端 timeout 总体限制，默认 sendTimeoutSeconds=600； 路由单独限制回调等待时间，dispatch／协商另有总等待截止时间。取消／超时后晚到结果不发送， 但不等于自动取消宿主正在运行的 LLM 或业务操作，宿主负责清理其资源。 同步回调入口应迅速返回，阻塞任务应交给异步执行器。 回调缺失、返回 null 或异常均明确失败；不确定发送失败不自动重发。 diff --git a/docs/zh/DESIGN.md b/docs/zh/DESIGN.md index 4c692fee..41cbe230 100644 --- a/docs/zh/DESIGN.md +++ b/docs/zh/DESIGN.md @@ -224,8 +224,9 @@ interface ControlPoint { } ``` -onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。 onSelfTask 返回本地 TaskResult；onRoute -选择允许的候选；onNegotiation 返回 Send 或 Stop。 未实现的回调明确失败，不回显成功、不选首分支、不自动同意。 +onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute +对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， +不回显成功、不默认放行条件边、不自动同意。 字段与完整示例见 [业务回调集成契约](BUSINESS_CALLBACKS.md)。 ## 5. A2A-T 扩展模型 @@ -252,11 +253,12 @@ Authorization-T 和 Notification-T 是独立业务操作，不属于 DAG。宿 步骤的 `next` 列表持有 `JumpCondition(step, condition)` 条目。路由规则： - **无 `next`** — 终端步骤，完成该分支。 -- **所有条件为空** — 无条件扇出：并行下发每个非终端下一步。 -- **有条件** — 条件路由：调用 `ControlPoint.onRoute`，返回单个 `RouteDecision.nextStep`。 - 引擎强制要求返回的步骤在声明的条件中；无效返回值会使工作流失败并报告错误。 +- **空条件** — 无条件边：不调用 `ControlPoint.onRoute`，直接激活。null、空字符串和纯空白都按空条件处理。 +- **非空条件** — 条件边：针对该边调用一次 `ControlPoint.onRoute`，仅当返回的 `RouteDecision.allowed` 为 true 时激活。 -这使得条件分支是 N 选 1 选择，无条件扇出是自动并行下发。 +所有条件边独立判断并可并发执行。引擎等待全部判断结束后，统一激活所有无条件边和判断为 true 的条件边；多个 true 会形成并行扇出。 +全部为 false 且没有无条件边时，当前分支正常结束。回调失败、超时或返回 null 时，任何后继节点都不会下发，工作流明确失败。 +路由失败信息会标明源节点和目标节点，并保留原始异常原因。同一源节点中指向相同目标的重复边会在图校验阶段被拒绝。 --- diff --git a/docs/zh/DEVELOPER_GUIDE.md b/docs/zh/DEVELOPER_GUIDE.md index 4a4a5082..be5fbf01 100644 --- a/docs/zh/DEVELOPER_GUIDE.md +++ b/docs/zh/DEVELOPER_GUIDE.md @@ -35,8 +35,9 @@ interface ControlPoint { } ``` -onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。 onSelfTask 返回本地 TaskResult；onRoute -选择允许的候选；onNegotiation 返回 Send 或 Stop。 未实现的回调明确失败，不回显成功、不选首分支、不自动同意。 +onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute +对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， +不回显成功、不默认放行条件边、不自动同意。 字段与完整示例见 [业务回调集成契约](BUSINESS_CALLBACKS.md)。 ```java @@ -46,8 +47,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); @@ -81,26 +82,26 @@ ExecutionResult result = ExecutePsop.builder() 事件来自三层：运行器（生命周期括号）、执行器（步骤/任务/路由）、引擎客户端（智能体流量、协商）。 -| 事件 | 层级 | 触发时机 | 关键数据 | -|-------------------------|----------------|----------------------------------------------------|-------------------------------------------------------| -| `start` | 运行器 | 工作流开始 | `workflow`、`steps` | -| `step_start` | 执行器 | 步骤开始 | `step` | -| `task_request` | 执行器 | 子任务下发到 `onTask`/`onSelfTask` | `step`、`agent`、`task` | -| `task_response` | 执行器 | 远端任务完成或 onSelfTask 返回 TaskResult | `step`、`agent`、`task`、`outputs` | -| `task_status_changed` | 执行器 | 任务状态变更（pending → running → success/failed） | `step`、`agent`、`task`、`status` | -| `route_decision` | 执行器 | 分支选择 | `step`、`next`、`reason` | -| `step_complete` | 执行器 | 步骤完成 | `step`、`results` | -| `workflow_complete` | 执行器 | 所有步骤完成 | `history`、`step_outputs` | -| `agent_request` | engine client | 准备下发（不是 wire 日志） | `agent`, `content` | -| `agent_response` | engine client | 远端响应已组装 | `agent`, `response`, `receivedMessages` | -| `agent_status_update` | 引擎客户端 | 智能体 SSE 状态更新 | `agent`、`state`、`is_final` | -| `agent_artifact_update` | 引擎客户端 | 智能体 SSE 产物更新 | `agent`、`artifact_name`、`text` | -| `negotiation_request` | engine client | 有效 Propose 进入业务接管 | `agent`, `request`, `exchange` | -| `negotiation_resolved` | engine client | 宿主 Send 通过关联检查；不等于任务成功 | `agent`, `reply`, `exchange` | -| `negotiation_failed` | engine client | 本地协商交互失败 | `agent`, `exchange`, `errorType` | -| `complete` | 运行器 | 工作流成功 | `history`、`step_outputs` | -| `error` | 运行器或执行器 | 工作流失败 | 运行器：`error`、`history`；执行器：`step`、`results` | -| `close` | 运行器 | 清理完成 | （空） | +| 事件 | 层级 | 触发时机 | 关键数据 | +|-------------------------|----------------|----------------------------------------------------|-----------------------------------------------------------------| +| `start` | 运行器 | 工作流开始 | `workflow`、`steps` | +| `step_start` | 执行器 | 步骤开始 | `step` | +| `task_request` | 执行器 | 子任务下发到 `onTask`/`onSelfTask` | `step`、`agent`、`task` | +| `task_response` | 执行器 | 远端任务完成或 onSelfTask 返回 TaskResult | `step`、`agent`、`task`、`outputs` | +| `task_status_changed` | 执行器 | 任务状态变更（pending → running → success/failed） | `step`、`agent`、`task`、`status` | +| `route_decision` | 执行器 | 路由阶段成功后记录一条出边的判断结果 | `step`、`next`、`condition`、`conditional`、`allowed`、`reason` | +| `step_complete` | 执行器 | 步骤完成 | `step`、`results` | +| `workflow_complete` | 执行器 | 所有步骤完成 | `history`、`step_outputs` | +| `agent_request` | engine client | 准备下发（不是 wire 日志） | `agent`, `content` | +| `agent_response` | engine client | 远端响应已组装 | `agent`, `response`, `receivedMessages` | +| `agent_status_update` | 引擎客户端 | 智能体 SSE 状态更新 | `agent`、`state`、`is_final` | +| `agent_artifact_update` | 引擎客户端 | 智能体 SSE 产物更新 | `agent`、`artifact_name`、`text` | +| `negotiation_request` | engine client | 有效 Propose 进入业务接管 | `agent`, `request`, `exchange` | +| `negotiation_resolved` | engine client | 宿主 Send 通过关联检查；不等于任务成功 | `agent`, `reply`, `exchange` | +| `negotiation_failed` | engine client | 本地协商交互失败 | `agent`, `exchange`, `errorType` | +| `complete` | 运行器 | 工作流成功 | `history`、`step_outputs` | +| `error` | 运行器或执行器 | 工作流失败 | 运行器：`error`、`history`；执行器：`step`、`results` | +| `close` | 运行器 | 清理完成 | （空） | ## 6. 中间层（Layer 1: WorkflowExecutor） diff --git a/docs/zh/INTEGRATION_GUIDE.md b/docs/zh/INTEGRATION_GUIDE.md index 0dde4252..dc9e60c8 100644 --- a/docs/zh/INTEGRATION_GUIDE.md +++ b/docs/zh/INTEGRATION_GUIDE.md @@ -30,9 +30,8 @@ A2A-T 工作流执行引擎是一个 Java SDK，用于基于 A2A 协议和 A2A-T 完整可运行源码：[HostQuickStart.java](../../samples/src/main/java/dev/openan/workflow/engine/examples/demo/HostQuickStart.java)。 该源码参与编译，远端任务→本地汇总流程由 HostQuickStartTest 验证。 -可复制到集成方工程，或在 IDEA 的 samples 模块运行，传入注册中心 URL、目标 AgentCard 名称、凭证路径。 -下列片段解释同一套 API；AgentCard 加载方式二选一，Task-T 内容生成需要由业务实现。 -非空业务条件必须实现 onRoute 策略；这里使用无条件边，不依赖默认选路。 +可复制到集成方工程，或在 IDEA 的 samples 模块运行，传入注册中心 URL、目标 AgentCard 名称、凭证路径。 下列片段解释同一套 +API；AgentCard 加载方式二选一，Task-T 内容生成需要由业务实现。 每条非空业务条件都需要 onRoute 独立判断；这里使用无条件边，不依赖默认选路。 ### 4.1 定义工作流 @@ -109,8 +108,9 @@ interface ControlPoint { } ``` -onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。 onSelfTask 返回本地 TaskResult；onRoute -选择允许的候选；onNegotiation 返回 Send 或 Stop。 未实现的回调明确失败，不回显成功、不选首分支、不自动同意。 +onTask 返回最终 parts/metadata/extensions，引擎封装发送，不再生成或改写内容。onSelfTask 返回本地 TaskResult；onRoute +对一条条件边独立返回放行或拒绝；onNegotiation 返回 Send 或 Stop。无条件边不调用 onRoute 并始终放行。未实现的回调明确失败， +不回显成功、不默认放行条件边、不自动同意。 字段与完整示例见 [业务回调集成契约](BUSINESS_CALLBACKS.md)。 ```java @@ -120,8 +120,8 @@ ControlPoint callbacks = ControlPoint.builder() .onSelfTask(request -> CompletableFuture.completedFuture( TaskResult.success(List.of(Map.of( "sourceResults", request.getWorkflowInput().upstreamResults()))))) - .onRoute(request -> CompletableFuture.failedFuture( - new IllegalStateException("Supply a routing policy for " + request.stepName()))) + .onRoute(request -> CompletableFuture.completedFuture( + RouteDecision.deny("Replace with host condition evaluation"))) .onNegotiation(request -> CompletableFuture.completedFuture( new NegotiationReply.Stop("manual.required", "Manual confirmation required"))) .build(); diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java index e05dd50c..e091678d 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchControlPoint.java @@ -41,9 +41,9 @@ /** * ControlPoint for the SPN cross-city diagnosis workflow. * - *

Handles task dispatch (with city-specific message enrichment), route decisions (fault-based - * routing to recovery steps), and negotiation responses. Authorization-T and Notification-T are - * initiated independently of the workflow, not handled here. + *

Handles task content, local aggregation, and negotiation responses. The sample workflow uses + * only unconditional edges, so it does not define a conditional routing policy. Authorization-T and + * Notification-T are initiated independently of the workflow, not handled here. * *

SRP: this class only contains workflow decision logic, separating it from the agent executor * that handles message I/O. @@ -127,7 +127,8 @@ public CompletableFuture onTask(TaskRequest request) { .get(dev.openan.workflow.engine.client.A2ATExtension.TASK_T.uri()) .toString())); log.info( - "[onTask] DEMO_NEGOTIATION agent={}, fault=missing-port, source=explicit-sample-switch", + "[onTask] DEMO_NEGOTIATION agent={}, fault=missing-port," + + " source=explicit-sample-switch", request.getAgentName()); return new MessageContent(content.parts(), metadata, content.extensions()); } @@ -161,10 +162,8 @@ public CompletableFuture onSelfTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest request) { - // merge_analysis has an unconditional next -> endNode, so the executor - // never calls onRoute for it. Recovery is self-triggered by SPN agents - // via the active Authorization-T whitelist and reported through - // the Notification-T channel. Just delegate to the default routing. + // The sample PSOP has only unconditional edges, which bypass this callback. Fail fast if an + // orchestration center supplies a conditional edge without a matching host business policy. return super.onRoute(request); } diff --git a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java index e518fd2a..d5d55c00 100644 --- a/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java +++ b/samples/src/main/java/dev/openan/workflow/engine/examples/workbench/WorkbenchOrchestrator.java @@ -150,7 +150,11 @@ private static String outputText(Map output) { } return output.values().stream() .filter(value -> value != null && !String.valueOf(value).isBlank()) - .flatMap(value -> value instanceof List values ? values.stream() : java.util.stream.Stream.of(value)) + .flatMap( + value -> + value instanceof List values + ? values.stream() + : java.util.stream.Stream.of(value)) .map(WorkbenchOrchestrator::renderOutput) .collect(java.util.stream.Collectors.joining("\n\n")); } @@ -166,13 +170,18 @@ private static String renderOutput(Object value) { static String buildResultText(ExecutionResult result, Workflow workflow) { if (result.isSuccess() && result.getStepOutputs() != null) { - String terminalOutputs = workflow.getSteps().stream() - .filter(step -> result.getStepOutputs().containsKey(step.getName())) - .filter(step -> step.getNext() == null || step.getNext().stream() - .noneMatch(next -> result.getStepOutputs().containsKey(next.getStep()))) - .map(step -> outputText(result.getStepOutputs().get(step.getName()))) - .filter(output -> !output.isBlank()) - .collect(java.util.stream.Collectors.joining("\n\n")); + String terminalOutputs = + workflow.getSteps().stream() + .filter(step -> result.getStepOutputs().containsKey(step.getName())) + .filter( + step -> + step.getNext() == null + || step.getNext().stream() + .noneMatch( + next -> result.getStepOutputs().containsKey(next.getStep()))) + .map(step -> outputText(result.getStepOutputs().get(step.getName()))) + .filter(output -> !output.isBlank()) + .collect(java.util.stream.Collectors.joining("\n\n")); if (!terminalOutputs.isBlank()) return terminalOutputs; } return buildResultText(result); @@ -390,7 +399,8 @@ public void onEvent(String type, Map data) { case EventType.TASK_REQUEST -> log.info(" [TASK_REQUEST] agent={}", data.get("agent")); case EventType.TASK_RESPONSE -> log.info( - " [TASK_RESPONSE] contextId={}, executionId={}, step={}, taskId={}, agent={}, success={}, errorCode={}", + " [TASK_RESPONSE] contextId={}, executionId={}, step={}, taskId={}, agent={}," + + " success={}, errorCode={}", contextId, data.get("executionId"), data.get("step"), @@ -416,7 +426,13 @@ public void onEvent(String type, Map data) { data.get("text") != null ? ((String) data.get("text")).length() : 0); case EventType.STEP_COMPLETE -> log.info(" [STEP_COMPLETE] {}", data.get("step")); case EventType.ROUTE_DECISION -> - log.info(" [ROUTE] {} -> {}", data.get("step"), data.get("next")); + log.info( + " [ROUTE] {} -> {}, conditional={}, allowed={}, reason={}", + data.get("step"), + data.get("next"), + data.get("conditional"), + data.get("allowed"), + data.get("reason")); case EventType.COMPLETE -> log.info(" [COMPLETE]"); case EventType.ERROR -> log.error( diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/ControlPoint.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/ControlPoint.java index 69c2dd35..21b510d2 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/control/ControlPoint.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/control/ControlPoint.java @@ -19,11 +19,20 @@ package dev.openan.workflow.engine.control; -import dev.openan.workflow.engine.model.*; +import dev.openan.workflow.engine.model.MessageContent; +import dev.openan.workflow.engine.model.NegotiationReply; +import dev.openan.workflow.engine.model.NegotiationRequest; +import dev.openan.workflow.engine.model.RouteDecision; +import dev.openan.workflow.engine.model.RouteRequest; +import dev.openan.workflow.engine.model.TaskRequest; +import dev.openan.workflow.engine.model.TaskResult; import java.util.concurrent.CompletableFuture; import java.util.function.Function; -/** Business-only callbacks. Implementations must support concurrent calls from different tasks. */ +/** + * Business-only callbacks. Implementations must support concurrent calls from different tasks and + * from different conditional edges of the same workflow step. + */ public interface ControlPoint { /** Registers only the capabilities the host needs. */ static Builder builder() { @@ -42,10 +51,14 @@ default CompletableFuture onSelfTask(TaskRequest request) { new IllegalStateException("onSelfTask handler is required for " + request.getStepName())); } - /** Selects a permitted branch. Unconditional edges bypass this callback. */ + /** Evaluates one conditional edge. Unconditional edges bypass this callback and always run. */ default CompletableFuture onRoute(RouteRequest request) { return CompletableFuture.failedFuture( - new IllegalStateException("onRoute handler is required for " + request.stepName())); + new IllegalStateException( + "onRoute handler is required for edge " + + request.stepName() + + " -> " + + request.nextStep())); } /** Answers the proposal. Missing handlers never implicitly consent. */ 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 def3deaa..15ea3604 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 @@ -25,6 +25,8 @@ import dev.openan.workflow.engine.control.EventType; import dev.openan.workflow.engine.model.ExecutionResult; import dev.openan.workflow.engine.model.JumpCondition; +import dev.openan.workflow.engine.model.RouteDecision; +import dev.openan.workflow.engine.model.RouteRequest; import dev.openan.workflow.engine.model.StepType; import dev.openan.workflow.engine.model.Task; import dev.openan.workflow.engine.model.TaskExecutionResult; @@ -39,13 +41,17 @@ import java.util.Collections; import java.util.Deque; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; import org.slf4j.Logger; @@ -169,87 +175,8 @@ private void emit(String type, Map data) { } } - public CompletableFuture run() { - if (!started.compareAndSet(false, true)) - return CompletableFuture.failedFuture( - new IllegalStateException("WorkflowExecutor is single-use")); - // NOTE: START lifecycle event is emitted by the runner (ExecutePsop), - // not here. Mirrors Python SDK where the executor emits only - // step/task/route events and the runner emits start/complete/error/close. - log.info( - "[Executor] Starting workflow: {} ({} steps)", - workflow.getName(), - workflow.getSteps().size()); - try { - validateWorkflowGraph(); - } catch (IllegalArgumentException e) { - log.error("[Executor] Invalid workflow graph: {}", e.getMessage()); - emit(EventType.ERROR, Map.of("error", e.getMessage())); - return CompletableFuture.completedFuture( - ExecutionResult.builder() - .success(false) - .history(new ArrayList<>(executionHistory)) - .stepOutputs(new HashMap<>(stepOutputs)) - .error(e.getMessage()) - .build()); - } - Deque pending = new ConcurrentLinkedDeque<>(); - Set scheduled = ConcurrentHashMap.newKeySet(); - Set activated = ConcurrentHashMap.newKeySet(); - for (int i = 0; i < workflow.getSteps().size(); i++) { - var s = workflow.getSteps().get(i); - if (contextBuilder.getStepPredecessors(s.getName()).isEmpty()) { - pending.add(i); - scheduled.add(i); - activated.add(i); - } - } - Set executed = ConcurrentHashMap.newKeySet(); - AtomicBoolean failed = new AtomicBoolean(); - CompletableFuture execution = - executeSteps(pending, scheduled, executed, activated, failed) - .thenApply( - v -> { - emit(EventType.WORKFLOW_COMPLETE, Map.of("success", !failed.get())); - log.info( - "[Executor] WORKFLOW_FINISHED executionId={}, workflow={}, success={}, tasks={}", - executionId, - workflow.getName(), - !failed.get(), - executionHistory.size()); - if (failed.get()) { - log.warn( - "[Executor] WORKFLOW_STOPPED executionId={}, reason=step_failed, skippedSteps={}", - executionId, - java.util.stream.IntStream.range(0, workflow.getSteps().size()) - .filter(index -> !executed.contains(index)) - .mapToObj(index -> workflow.getSteps().get(index).getName()) - .toList()); - } - return ExecutionResult.builder() - .success(!failed.get()) - .history(new ArrayList<>(executionHistory)) - .stepOutputs(new HashMap<>(stepOutputs)) - .error(failed.get() ? "Step execution failed" : null) - .build(); - }) - .exceptionally( - e -> { - log.error("[Executor] DAG traversal error: {}", e.getMessage(), e); - emit(EventType.ERROR, Map.of("error", e.getMessage())); - return ExecutionResult.builder() - .success(false) - .history(new ArrayList<>(executionHistory)) - .stepOutputs(new HashMap<>(stepOutputs)) - .error(e.getMessage()) - .build(); - }); - execution.whenComplete( - (value, error) -> { - stopped.set(true); - activeTasks.forEach(task -> task.cancel(true)); - }); - return execution; + private static boolean isUnconditional(JumpCondition edge) { + return edge.getCondition() == null || edge.getCondition().isBlank(); } private CompletableFuture executeSteps( @@ -328,111 +255,26 @@ private CompletableFuture executeSteps( .thenCompose(v -> executeSteps(pending, scheduled, executed, activated, failed)); } - private CompletableFuture executeStep( - WorkflowStep step, - Set scheduled, - Set activated, - Deque pending, - AtomicBoolean failed) { - emit(EventType.STEP_START, Map.of("step", step.getName())); - log.info("--- Executing step: {} ---", step.getName()); - return executeSubtasks(step) - .thenCompose( - result -> { - stepOutputs.put(step.getName(), result.results()); - stepExecutionResults.put(step.getName(), result.taskResults()); - if (!result.success()) { - log.warn( - "[Executor] STEP_FAILED executionId={}, step={}, action=stop-downstream", - executionId, - step.getName()); - emit( - EventType.ERROR, - Map.of( - "step", - step.getName(), - "results", - result.results(), - "error", - "Step execution failed", - "errorCode", - "workflow.step_failed")); - failed.set(true); - return CompletableFuture.completedFuture(null); - } - emit( - EventType.STEP_COMPLETE, - Map.of("step", step.getName(), "results", result.results())); - return determineNextSteps(step, result.results()) - .thenAccept( - nextIndices -> { - for (int i = nextIndices.size() - 1; i >= 0; i--) { - int nxt = nextIndices.get(i); - activated.add(nxt); - if (scheduled.add(nxt)) { - pending.addFirst(nxt); - } - } - }); - }); + private static IllegalStateException routeEvaluationFailure( + WorkflowStep step, JumpCondition edge, Throwable error) { + Throwable cause = unwrapCompletionException(error); + if (cause instanceof RouteEvaluationException routeError) { + return routeError; + } + return new RouteEvaluationException( + "onRoute failed for edge " + + step.getName() + + " -> " + + edge.getStep() + + ": " + + failureMessage(cause), + cause); } - private void validateWorkflowGraph() { - Map indices = new HashMap<>(); - for (int i = 0; i < workflow.getSteps().size(); i++) { - String name = workflow.getSteps().get(i).getName(); - 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); - } - } - List> graph = new ArrayList<>(); - for (int i = 0; i < workflow.getSteps().size(); i++) graph.add(new ArrayList<>()); - for (int i = 0; i < workflow.getSteps().size(); i++) { - WorkflowStep step = workflow.getSteps().get(i); - if (step.getNext() == null) continue; - for (JumpCondition jump : step.getNext()) { - if (isTerminalRoute(jump.getStep())) continue; - Integer target = indices.get(jump.getStep()); - if (target == null) { - throw new IllegalArgumentException( - "Step " + step.getName() + " references missing step " + jump.getStep()); - } - graph.get(i).add(target); - } - } - for (WorkflowStep step : workflow.getSteps()) { - List contextFrom = step.getContextFrom(); - if (contextFrom == null || contextFrom.isEmpty()) continue; - if (contextFrom.contains("*") && contextFrom.size() != 1) { - throw new IllegalArgumentException( - "Step " + step.getName() + " cannot combine '*' with named context sources"); - } - if (contextFrom.contains("*")) continue; - Set ancestors = Set.copyOf(contextBuilder.getAllPredecessors(step.getName())); - for (String source : contextFrom) { - if (!indices.containsKey(source)) { - throw new IllegalArgumentException( - "Step " + step.getName() + " references missing context source " + source); - } - if (!ancestors.contains(source)) { - throw new IllegalArgumentException( - "Step " - + step.getName() - + " context source " - + source - + " is not an upstream dependency"); - } - } - } - int[] state = new int[workflow.getSteps().size()]; - for (int i = 0; i < state.length; i++) detectCycle(i, graph, state); + private static String failureMessage(Throwable error) { + Throwable cause = unwrapCompletionException(error); + String message = cause.getMessage(); + return message == null || message.isBlank() ? cause.getClass().getSimpleName() : message; } private void detectCycle(int node, List> graph, int[] state) { @@ -582,117 +424,27 @@ private StepResult collectAllSuccess(List> futures collectExecutionResults(completedResults)); } - private StepResult processTaskResult( - WorkflowStep step, Task task, int subtaskIndex, TaskResult response) { - task.setStatus(response.isSuccess() ? TaskStatus.SUCCESS : TaskStatus.FAILED); - emit( - EventType.TASK_STATUS_CHANGED, - Map.of( - "step", - step.getName(), - "subtask_index", - subtaskIndex, - "agent", - task.getAgent(), - "status", - task.getStatus().getValue())); - String status = response.isSuccess() ? "success" : "failed"; - log.info("[Executor] Task {} -> {}: {}", task.getDescription(), task.getAgent(), status); - if (!response.isSuccess()) { - log.warn( - "[Executor] TASK_FAILED executionId={}, step={}, taskId={}, agent={}, errorCode={}, reason={}", - executionId, - step.getName(), - taskId(step.getName(), subtaskIndex), - task.getAgent(), - response.getErrorCode(), - SensitiveDataRedactor.redact(response.getError()) - .replace("\r", "\\r") - .replace("\n", "\\n")); + private static Throwable unwrapCompletionException(Throwable error) { + Throwable current = Objects.requireNonNull(error, "error"); + while (current instanceof CompletionException && current.getCause() != null) { + current = current.getCause(); } - if (response.isSuccess() && !response.getOutputs().isEmpty()) { - log.debug("[Executor] Task outputs from {}: [{}]", task.getAgent(), response.getOutputs()); - } - List outputs = response.getOutputs(); - executionHistory.add( - Map.of( - "step", - step.getName(), - "subtask_index", - subtaskIndex, - "task", - task.getDescription(), - "agent", - task.getAgent(), - "status", - status, - "taskId", - taskId(step.getName(), subtaskIndex), - "outputs", - outputs, - "error", - response.getError() == null ? "" : response.getError(), - "errorCode", - response.getErrorCode() == null ? "" : response.getErrorCode(), - "errorDetails", - response.getErrorDetails())); - emit( - EventType.TASK_RESPONSE, - Map.of( - "step", - step.getName(), - "subtask_index", - subtaskIndex, - "agent", - task.getAgent(), - "task", - task.getDescription(), - "outputs", - outputs, - "success", - response.isSuccess(), - "taskId", - taskId(step.getName(), subtaskIndex), - "error", - response.getError() == null ? "" : response.getError(), - "errorCode", - response.getErrorCode() == null ? "" : response.getErrorCode(), - "errorDetails", - response.getErrorDetails())); - return new StepResult( - task.getDescription(), - task.getAgent(), - subtaskIndex, - outputs, - response.isSuccess(), - null, - List.of( - new TaskExecutionResult( - task.getAgent(), - task.getSkill(), - taskId(step.getName(), subtaskIndex), - task.getDescription(), - task.getStatus(), - outputs, - response.getReceivedMessages(), - response.getError(), - response.getErrorCode(), - response.getErrorDetails()))); - } - - private StepResult processTaskError(WorkflowStep step, Task task, int subtaskIndex, Throwable e) { - return processTaskResult(step, task, subtaskIndex, FailureMapping.from(e)); - } - - /** - * ANY_SUCCESS logic: iterate futures as they complete; on the first success, cancel the rest and - * return success=true. If all fail, return success=false. Mirrors Python's asyncio.as_completed - * loop. - */ - private CompletableFuture anySuccess(List> futures) { - if (futures.isEmpty()) { - return CompletableFuture.completedFuture( - new StepResult(null, null, -1, List.of(), true, Map.of(), List.of())); + return current; + } + + private StepResult processTaskError(WorkflowStep step, Task task, int subtaskIndex, Throwable e) { + return processTaskResult(step, task, subtaskIndex, FailureMapping.from(e)); + } + + /** + * ANY_SUCCESS logic: iterate futures as they complete; on the first success, cancel the rest and + * return success=true. If all fail, return success=false. Mirrors Python's asyncio.as_completed + * loop. + */ + private CompletableFuture anySuccess(List> futures) { + if (futures.isEmpty()) { + return CompletableFuture.completedFuture( + new StepResult(null, null, -1, List.of(), true, Map.of(), List.of())); } CompletableFuture result = new CompletableFuture<>(); int total = futures.size(); @@ -772,79 +524,428 @@ private CompletableFuture anySuccess(List> determineNextSteps( - WorkflowStep step, Map stepResult) { - if (step.getNext() == null || step.getNext().isEmpty()) { - return CompletableFuture.completedFuture(List.of()); + public CompletableFuture run() { + if (!started.compareAndSet(false, true)) + return CompletableFuture.failedFuture( + new IllegalStateException("WorkflowExecutor is single-use")); + // NOTE: START lifecycle event is emitted by the runner (ExecutePsop), + // not here. Mirrors Python SDK where the executor emits only + // step/task/route events and the runner emits start/complete/error/close. + log.info( + "[Executor] Starting workflow: {} ({} steps)", + workflow.getName(), + workflow.getSteps().size()); + try { + validateWorkflowGraph(); + } catch (IllegalArgumentException e) { + log.error("[Executor] Invalid workflow graph: {}", e.getMessage()); + emit(EventType.ERROR, Map.of("error", e.getMessage())); + return CompletableFuture.completedFuture( + ExecutionResult.builder() + .success(false) + .history(new ArrayList<>(executionHistory)) + .stepOutputs(new HashMap<>(stepOutputs)) + .error(e.getMessage()) + .build()); + } + Deque pending = new ConcurrentLinkedDeque<>(); + Set scheduled = ConcurrentHashMap.newKeySet(); + Set activated = ConcurrentHashMap.newKeySet(); + for (int i = 0; i < workflow.getSteps().size(); i++) { + var s = workflow.getSteps().get(i); + if (contextBuilder.getStepPredecessors(s.getName()).isEmpty()) { + pending.add(i); + scheduled.add(i); + activated.add(i); + } } - if (step.getNext().stream() - .allMatch(jc -> jc.getCondition() == null || jc.getCondition().isEmpty())) { - List indices = new ArrayList<>(); - for (var jc : step.getNext()) { - if (isTerminalRoute(jc.getStep())) { - continue; + Set executed = ConcurrentHashMap.newKeySet(); + AtomicBoolean failed = new AtomicBoolean(); + CompletableFuture execution = + executeSteps(pending, scheduled, executed, activated, failed) + .thenApply( + v -> { + emit(EventType.WORKFLOW_COMPLETE, Map.of("success", !failed.get())); + log.info( + "[Executor] WORKFLOW_FINISHED executionId={}, workflow={}, success={}," + + " tasks={}", + executionId, + workflow.getName(), + !failed.get(), + executionHistory.size()); + if (failed.get()) { + log.warn( + "[Executor] WORKFLOW_STOPPED executionId={}, reason=step_failed," + + " skippedSteps={}", + executionId, + java.util.stream.IntStream.range(0, workflow.getSteps().size()) + .filter(index -> !executed.contains(index)) + .mapToObj(index -> workflow.getSteps().get(index).getName()) + .toList()); + } + return ExecutionResult.builder() + .success(!failed.get()) + .history(new ArrayList<>(executionHistory)) + .stepOutputs(new HashMap<>(stepOutputs)) + .error(failed.get() ? "Step execution failed" : null) + .build(); + }) + .exceptionally( + e -> { + Throwable failure = unwrapCompletionException(e); + String message = failureMessage(failure); + log.error("[Executor] DAG traversal error: {}", message, failure); + emit(EventType.ERROR, Map.of("error", message)); + return ExecutionResult.builder() + .success(false) + .history(new ArrayList<>(executionHistory)) + .stepOutputs(new HashMap<>(stepOutputs)) + .error(message) + .build(); + }); + execution.whenComplete( + (value, error) -> { + stopped.set(true); + activeTasks.forEach(task -> task.cancel(true)); + }); + return execution; + } + + private CompletableFuture executeStep( + WorkflowStep step, + Set scheduled, + Set activated, + Deque pending, + AtomicBoolean failed) { + emit(EventType.STEP_START, Map.of("step", step.getName())); + log.info("--- Executing step: {} ---", step.getName()); + return executeSubtasks(step) + .thenCompose( + result -> { + stepOutputs.put(step.getName(), result.results()); + stepExecutionResults.put(step.getName(), result.taskResults()); + if (!result.success()) { + log.warn( + "[Executor] STEP_FAILED executionId={}, step={}, action=stop-downstream", + executionId, + step.getName()); + emit( + EventType.ERROR, + Map.of( + "step", + step.getName(), + "results", + result.results(), + "error", + "Step execution failed", + "errorCode", + "workflow.step_failed")); + failed.set(true); + return CompletableFuture.completedFuture(null); + } + emit( + EventType.STEP_COMPLETE, + Map.of("step", step.getName(), "results", result.results())); + return determineNextSteps(step) + .thenAccept( + nextIndices -> { + for (int i = nextIndices.size() - 1; i >= 0; i--) { + int nxt = nextIndices.get(i); + activated.add(nxt); + if (scheduled.add(nxt)) { + pending.addFirst(nxt); + } + } + }); + }); + } + + private StepResult processTaskResult( + WorkflowStep step, Task task, int subtaskIndex, TaskResult response) { + task.setStatus(response.isSuccess() ? TaskStatus.SUCCESS : TaskStatus.FAILED); + emit( + EventType.TASK_STATUS_CHANGED, + Map.of( + "step", + step.getName(), + "subtask_index", + subtaskIndex, + "agent", + task.getAgent(), + "status", + task.getStatus().getValue())); + String status = response.isSuccess() ? "success" : "failed"; + log.info("[Executor] Task {} -> {}: {}", task.getDescription(), task.getAgent(), status); + if (!response.isSuccess()) { + log.warn( + "[Executor] TASK_FAILED executionId={}, step={}, taskId={}, agent={}, errorCode={}," + + " reason={}", + executionId, + step.getName(), + taskId(step.getName(), subtaskIndex), + task.getAgent(), + response.getErrorCode(), + SensitiveDataRedactor.redact(response.getError()) + .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( + "step", + step.getName(), + "subtask_index", + subtaskIndex, + "task", + task.getDescription(), + "agent", + task.getAgent(), + "status", + status, + "taskId", + taskId(step.getName(), subtaskIndex), + "outputs", + outputs, + "error", + response.getError() == null ? "" : response.getError(), + "errorCode", + response.getErrorCode() == null ? "" : response.getErrorCode(), + "errorDetails", + response.getErrorDetails())); + emit( + EventType.TASK_RESPONSE, + Map.of( + "step", + step.getName(), + "subtask_index", + subtaskIndex, + "agent", + task.getAgent(), + "task", + task.getDescription(), + "outputs", + outputs, + "success", + response.isSuccess(), + "taskId", + taskId(step.getName(), subtaskIndex), + "error", + response.getError() == null ? "" : response.getError(), + "errorCode", + response.getErrorCode() == null ? "" : response.getErrorCode(), + "errorDetails", + response.getErrorDetails())); + return new StepResult( + task.getDescription(), + task.getAgent(), + subtaskIndex, + outputs, + response.isSuccess(), + null, + List.of( + new TaskExecutionResult( + task.getAgent(), + task.getSkill(), + taskId(step.getName(), subtaskIndex), + task.getDescription(), + task.getStatus(), + outputs, + response.getReceivedMessages(), + response.getError(), + response.getErrorCode(), + response.getErrorDetails()))); + } + + private void validateWorkflowGraph() { + Map indices = new HashMap<>(); + for (int i = 0; i < workflow.getSteps().size(); i++) { + String name = workflow.getSteps().get(i).getName(); + 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); + } + } + List> graph = new ArrayList<>(); + for (int i = 0; i < workflow.getSteps().size(); i++) graph.add(new ArrayList<>()); + for (int i = 0; i < workflow.getSteps().size(); i++) { + WorkflowStep step = workflow.getSteps().get(i); + if (step.getNext() == null) continue; + Set targets = new HashSet<>(); + for (JumpCondition jump : step.getNext()) { + if (jump == null) { + throw new IllegalArgumentException( + "Step " + step.getName() + " contains a null outgoing edge"); } - Integer idx = contextBuilder.findStepIndex(jc.getStep()); - if (idx != null) { - indices.add(idx); + String nextStep = jump.getStep(); + if (nextStep == null || nextStep.isBlank()) { + throw new IllegalArgumentException( + "Step " + step.getName() + " contains an outgoing edge with a blank target"); } + if (!targets.add(nextStep)) { + throw new IllegalArgumentException( + "Step " + step.getName() + " contains duplicate outgoing target " + nextStep); + } + if (isTerminalRoute(nextStep)) continue; + Integer target = indices.get(nextStep); + if (target == null) { + throw new IllegalArgumentException( + "Step " + step.getName() + " references missing step " + nextStep); + } + graph.get(i).add(target); } - return CompletableFuture.completedFuture(indices); } - var routeRequest = - new dev.openan.workflow.engine.model.RouteRequest( + for (WorkflowStep step : workflow.getSteps()) { + List contextFrom = step.getContextFrom(); + if (contextFrom == null || contextFrom.isEmpty()) continue; + if (contextFrom.contains("*") && contextFrom.size() != 1) { + throw new IllegalArgumentException( + "Step " + step.getName() + " cannot combine '*' with named context sources"); + } + if (contextFrom.contains("*")) continue; + Set ancestors = Set.copyOf(contextBuilder.getAllPredecessors(step.getName())); + for (String source : contextFrom) { + if (!indices.containsKey(source)) { + throw new IllegalArgumentException( + "Step " + step.getName() + " references missing context source " + source); + } + if (!ancestors.contains(source)) { + throw new IllegalArgumentException( + "Step " + + step.getName() + + " context source " + + source + + " is not an upstream dependency"); + } + } + } + int[] state = new int[workflow.getSteps().size()]; + for (int i = 0; i < state.length; i++) detectCycle(i, graph, state); + } + + private CompletableFuture> determineNextSteps(WorkflowStep step) { + if (step.getNext() == null || step.getNext().isEmpty()) { + return CompletableFuture.completedFuture(List.of()); + } + var workflowInput = contextBuilder.buildWorkflowInput(step, stepExecutionResults); + var currentResults = stepExecutionResults.getOrDefault(step.getName(), List.of()); + List evaluations = + step.getNext().stream() + .map(edge -> evaluateRoute(step, edge, workflowInput, currentResults)) + .toList(); + CompletableFuture completed = + CompletableFuture.allOf( + evaluations.stream().map(RouteEvaluation::decision).toArray(CompletableFuture[]::new)); + return completed.thenApply(ignored -> collectAllowedTargets(step, evaluations)); + } + + private RouteEvaluation evaluateRoute( + WorkflowStep step, + JumpCondition edge, + WorkflowInput workflowInput, + List currentResults) { + if (isUnconditional(edge)) { + return new RouteEvaluation( + edge, + false, + CompletableFuture.completedFuture(RouteDecision.allow("unconditional edge"))); + } + var request = + new RouteRequest( executionId, step.getName(), - contextBuilder.buildWorkflowInput(step, stepExecutionResults), - stepExecutionResults.getOrDefault(step.getName(), List.of()), - step.getNext().stream() - .map( - option -> - new dev.openan.workflow.engine.model.RouteRequest.RouteOption( - option.getStep(), option.getCondition())) - .toList()); - 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(), - decision.getNextStep(), - decision.getReason()); - emit( - EventType.ROUTE_DECISION, - Map.of( - "step", - step.getName(), - "next", - decision.getNextStep(), - "reason", - decision.getReason())); - List allowed = - step.getNext().stream().map(JumpCondition::getStep).collect(Collectors.toList()); - if (!allowed.contains(decision.getNextStep())) { - throw new IllegalStateException( - "onRoute returned '" - + decision.getNextStep() - + "' for step '" - + step.getName() - + "', allowed next steps are " - + allowed); - } - if (isTerminalRoute(decision.getNextStep())) return List.of(); - Integer idx = contextBuilder.findStepIndex(decision.getNextStep()); - if (idx == null) - throw new IllegalStateException( - "Route target does not exist: " + decision.getNextStep()); - return List.of(idx); - }); + edge.getStep(), + edge.getCondition(), + workflowInput, + currentResults); + CompletableFuture decision; + try { + decision = controlPoint.onRoute(request); + } catch (RuntimeException error) { + decision = CompletableFuture.failedFuture(routeEvaluationFailure(step, edge, error)); + } + if (decision == null) { + decision = + CompletableFuture.failedFuture( + routeEvaluationFailure( + step, edge, new NullPointerException("onRoute returned null future"))); + } + 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; + })); + } + + private List collectAllowedTargets( + WorkflowStep step, List evaluations) { + List indices = new ArrayList<>(); + for (RouteEvaluation evaluation : evaluations) { + var decision = evaluation.decision().join(); + emitRouteDecision(step, evaluation, decision); + if (!decision.allowed() || isTerminalRoute(evaluation.edge().getStep())) continue; + Integer index = contextBuilder.findStepIndex(evaluation.edge().getStep()); + if (index == null) { + throw new IllegalStateException( + "Route target does not exist: " + evaluation.edge().getStep()); + } + indices.add(index); + } + return indices; } + private void emitRouteDecision( + WorkflowStep step, RouteEvaluation evaluation, RouteDecision decision) { + String condition = + evaluation.edge().getCondition() == null ? "" : evaluation.edge().getCondition(); + log.info( + "Route edge '{} -> {}': allowed={} ({})", + step.getName(), + evaluation.edge().getStep(), + decision.allowed(), + decision.reason()); + emit( + EventType.ROUTE_DECISION, + Map.of( + "step", step.getName(), + "next", evaluation.edge().getStep(), + "condition", condition, + "conditional", evaluation.conditional(), + "allowed", decision.allowed(), + "reason", decision.reason())); + } + + private static final class RouteEvaluationException extends IllegalStateException { + private RouteEvaluationException(String message, Throwable cause) { + super(message, cause); + } + } + + private record RouteEvaluation( + JumpCondition edge, boolean conditional, CompletableFuture decision) {} + private record StepResult( String taskDesc, String agentName, diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/JumpCondition.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/JumpCondition.java index a9619b64..42378fb9 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/JumpCondition.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/JumpCondition.java @@ -25,9 +25,9 @@ import lombok.NoArgsConstructor; /** - * One outgoing edge of a {@link WorkflowStep}: the next step name and a descriptive condition - * label offered to routing callbacks. Targets {@code end}, {@code retry}, and {@code endNode} - * are terminal routes, not step names. + * One outgoing edge of a {@link WorkflowStep}. A null, empty, or whitespace-only condition makes + * the edge unconditional; otherwise the engine offers this single edge to the routing callback. + * Targets {@code end}, {@code retry}, and {@code endNode} are terminal routes, not step names. */ @Data @NoArgsConstructor diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java index 0955ddcd..22fee55f 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteDecision.java @@ -19,21 +19,43 @@ package dev.openan.workflow.engine.model; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Data; -import lombok.NoArgsConstructor; - /** - * Chosen next step returned by an {@code onRoute} callback. {@code nextStep} must be one of the - * candidates offered by {@link RouteRequest}; any other value fails the workflow instead of - * being silently ignored. + * Business decision for one conditional outgoing edge. A denied edge is skipped; an allowed edge is + * activated together with every unconditional edge and every other allowed conditional edge. */ -@Data -@NoArgsConstructor -@AllArgsConstructor -@Builder -public class RouteDecision { - private String nextStep; - @Builder.Default private String reason = ""; +public record RouteDecision(boolean allowed, String reason) { + /** + * Creates a decision and normalizes a null reason to an empty string. + */ + public RouteDecision { + reason = reason == null ? "" : reason; + } + + /** + * Allows the edge without an explanatory reason. + */ + public static RouteDecision allow() { + return new RouteDecision(true, ""); + } + + /** + * Allows the edge and records the business reason for observability. + */ + public static RouteDecision allow(String reason) { + return new RouteDecision(true, reason); + } + + /** + * Denies the edge without an explanatory reason. + */ + public static RouteDecision deny() { + return new RouteDecision(false, ""); + } + + /** + * Denies the edge and records the business reason for observability. + */ + public static RouteDecision deny(String reason) { + return new RouteDecision(false, reason); + } } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java index 67449aca..400929fb 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/RouteRequest.java @@ -20,19 +20,30 @@ package dev.openan.workflow.engine.model; import java.util.List; +import java.util.Objects; -/** Conditional route input; upstream selection is identical to task callbacks. */ +/** + * Business input for evaluating one conditional outgoing edge. + */ public record RouteRequest( String executionId, String stepName, + String nextStep, + String condition, WorkflowInput workflowInput, - List currentResults, - List candidates) { + List currentResults) { + /** Creates and validates the immutable request for one nonblank conditional edge. */ public RouteRequest { + Objects.requireNonNull(executionId, "executionId"); + Objects.requireNonNull(stepName, "stepName"); + Objects.requireNonNull(nextStep, "nextStep"); + Objects.requireNonNull(condition, "condition"); + Objects.requireNonNull(workflowInput, "workflowInput"); + Objects.requireNonNull(currentResults, "currentResults"); + if (executionId.isBlank()) throw new IllegalArgumentException("executionId must not be blank"); + if (stepName.isBlank()) throw new IllegalArgumentException("stepName must not be blank"); + if (nextStep.isBlank()) throw new IllegalArgumentException("nextStep must not be blank"); + if (condition.isBlank()) throw new IllegalArgumentException("condition must not be blank"); currentResults = List.copyOf(currentResults); - candidates = List.copyOf(candidates); } - - /** A permitted target and its business condition. */ - public record RouteOption(String nextStep, String condition) {} } diff --git a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java index 601d9913..d522ee3e 100644 --- a/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java +++ b/workflow-engine/src/main/java/dev/openan/workflow/engine/model/WorkflowStep.java @@ -19,12 +19,13 @@ package dev.openan.workflow.engine.model; -import java.util.List; import lombok.AllArgsConstructor; import lombok.Builder; import lombok.Data; import lombok.NoArgsConstructor; +import java.util.List; + /** * One node of a {@link Workflow}: the subtasks to dispatch, the outgoing jumps, and the upstream * window its callbacks receive. @@ -34,14 +35,23 @@ @AllArgsConstructor @Builder public class WorkflowStep { - /** Unique step name; the terminal routes {@code end}, {@code retry}, {@code endNode} are reserved. */ + /** + * Unique step name; the terminal routes {@code end}, {@code retry}, {@code endNode} are reserved. + */ private String name; + /** Subtasks dispatched in parallel; completion policy is governed by {@code stepType}. */ @Builder.Default private List subtasks = List.of(); - /** Outgoing edges, evaluated after the step completes. */ + + /** + * Outgoing edges, evaluated after the step completes. Unconditional edges all run; each + * conditional edge is independently allowed or denied by the host. + */ @Builder.Default private List next = List.of(); + /** Layout hint retained for PSOP compatibility; not used by scheduling. */ @Builder.Default private int layer = 0; + /** * Selects which ancestor results the callbacks for this step receive. {@code null} selects the * direct predecessors, an empty list selects none, {@code "*"} selects all ancestors, and named @@ -49,10 +59,11 @@ public class WorkflowStep { * non-ancestor, fails validation before execution starts. */ private List contextFrom; + /** * Completion policy: {@code ALL_SUCCESS} advances only after every subtask succeeds, {@code - * ANY_SUCCESS} advances on the first success and cancels the rest, {@code SELF_LOOP} runs - * locally through {@code onSelfTask} with no agent network calls. + * ANY_SUCCESS} advances on the first success and cancels the rest, {@code SELF_LOOP} runs locally + * through {@code onSelfTask} with no agent network calls. */ @Builder.Default private StepType stepType = StepType.ALL_SUCCESS; } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java index eaf27ae9..47b23a44 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/control/DefaultControlPointTest.java @@ -18,11 +18,16 @@ */ package dev.openan.workflow.engine.control; -import static org.junit.jupiter.api.Assertions.*; +import dev.openan.workflow.engine.model.RouteRequest; +import dev.openan.workflow.engine.model.TaskRequest; +import dev.openan.workflow.engine.model.TaskResult; +import dev.openan.workflow.engine.model.WorkflowInput; +import org.junit.jupiter.api.Test; -import dev.openan.workflow.engine.model.*; import java.util.List; -import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; class DefaultControlPointTest { @@ -50,7 +55,8 @@ void conditionalRoutingRequiresBusinessHandler() { () -> new DefaultControlPoint() .onRoute( - new RouteRequest("run", "step", WorkflowInput.empty(), List.of(), List.of())) + new RouteRequest( + "run", "step", "next", "condition", WorkflowInput.empty(), List.of())) .join()); } diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/BusinessBoundaryTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/BusinessBoundaryTest.java index 7f756aaa..18e31f96 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/core/BusinessBoundaryTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/core/BusinessBoundaryTest.java @@ -19,13 +19,29 @@ package dev.openan.workflow.engine.core; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import dev.openan.workflow.engine.StubWorkflowEngineClient; -import dev.openan.workflow.engine.control.*; -import dev.openan.workflow.engine.model.*; -import java.util.*; -import java.util.concurrent.*; +import dev.openan.workflow.engine.control.ControlPoint; +import dev.openan.workflow.engine.model.JumpCondition; +import dev.openan.workflow.engine.model.RouteDecision; +import dev.openan.workflow.engine.model.RouteRequest; +import dev.openan.workflow.engine.model.StepType; +import dev.openan.workflow.engine.model.Task; +import dev.openan.workflow.engine.model.TaskRequest; +import dev.openan.workflow.engine.model.TaskResult; +import dev.openan.workflow.engine.model.TaskStatus; +import dev.openan.workflow.engine.model.Workflow; +import dev.openan.workflow.engine.model.WorkflowStep; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CopyOnWriteArrayList; import org.junit.jupiter.api.Test; class BusinessBoundaryTest { @@ -87,7 +103,7 @@ void routingUsesSelectedAncestorsAndCurrentTypedOutputsSeparately() { .onRoute( q -> { seen.add(q); - return CompletableFuture.completedFuture(new RouteDecision("endNode", "done")); + return CompletableFuture.completedFuture(RouteDecision.allow("done")); }) .build(); assertTrue( 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 e1bfc4b6..9bf2e162 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 @@ -19,16 +19,35 @@ package dev.openan.workflow.engine.core; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import dev.openan.workflow.engine.StubWorkflowEngineClient; import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.control.EventCallback; import dev.openan.workflow.engine.control.EventType; -import dev.openan.workflow.engine.model.*; +import dev.openan.workflow.engine.model.ExecutionResult; +import dev.openan.workflow.engine.model.JumpCondition; import dev.openan.workflow.engine.model.MessageContent; +import dev.openan.workflow.engine.model.RouteDecision; import dev.openan.workflow.engine.model.RouteRequest; -import java.util.*; +import dev.openan.workflow.engine.model.StepType; +import dev.openan.workflow.engine.model.Task; +import dev.openan.workflow.engine.model.TaskRequest; +import dev.openan.workflow.engine.model.TaskResult; +import dev.openan.workflow.engine.model.Workflow; +import dev.openan.workflow.engine.model.WorkflowStep; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; @@ -71,7 +90,7 @@ private JumpCondition jump(String step, String cond) { return JumpCondition.builder().step(step).condition(cond).build(); } - /** ControlPoint that auto-sends to the stub client and picks the first branch. */ + /** ControlPoint that prepares task content and allows every conditional edge. */ private ControlPoint autoCp() { return new ControlPoint() { @Override @@ -81,15 +100,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).reason("first").build()); + return CompletableFuture.completedFuture(RouteDecision.allow("test")); } }; } @@ -224,18 +235,7 @@ public CompletableFuture onSelfTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder() - .nextStep(conditions.get(0).getStep()) - .reason("first") - .build()); + return CompletableFuture.completedFuture(RouteDecision.allow("test")); } }; EventCallback callback = @@ -296,15 +296,10 @@ public CompletableFuture onTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("s3").reason("chose s3").build()); + routeRequest.nextStep().equals("s3") + ? RouteDecision.allow("chose s3") + : RouteDecision.deny("not selected")); } }; WorkflowExecutor exec = new WorkflowExecutor(wf, cp, stub, recordingCallback(), "", "zh"); @@ -317,7 +312,239 @@ public CompletableFuture onRoute(RouteRequest routeRequest) { } @Test - void onRouteInvalidStepFailsWorkflow() { + void mixedOutgoingEdgesRunUnconditionalAndAllowedConditionalTargets() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next( + List.of( + jump("always", ""), + jump("matched", "matched condition"), + jump("alsoMatched", "another match"), + jump("denied", "denied condition"))) + .build(); + List targets = + List.of( + WorkflowStep.builder().name("always").subtasks(List.of(task("B", "always"))).build(), + WorkflowStep.builder().name("matched").subtasks(List.of(task("C", "matched"))).build(), + WorkflowStep.builder() + .name("alsoMatched") + .subtasks(List.of(task("D", "also matched"))) + .build(), + WorkflowStep.builder().name("denied").subtasks(List.of(task("E", "denied"))).build()); + List requests = Collections.synchronizedList(new ArrayList<>()); + List> routeEvents = Collections.synchronizedList(new ArrayList<>()); + EventCallback routeEventCallback = + new EventCallback() { + @Override + public void onEvent(String type, Map data) { + if (EventType.ROUTE_DECISION.equals(type)) { + routeEvents.add(Map.copyOf(data)); + } + } + }; + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute( + request -> { + requests.add(request); + return CompletableFuture.completedFuture( + request.nextStep().equals("denied") + ? RouteDecision.deny("not applicable") + : RouteDecision.allow("matched")); + }) + .build(); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder() + .name("mixed-routes") + .steps( + java.util.stream.Stream.concat( + java.util.stream.Stream.of(start), targets.stream()) + .toList()) + .build(), + callbacks, + new StubWorkflowEngineClient("A", "B", "C", "D", "E"), + routeEventCallback, + "intent", + "zh") + .run() + .join(); + + assertTrue(result.isSuccess()); + assertEquals( + Set.of("matched", "alsoMatched", "denied"), + requests.stream().map(RouteRequest::nextStep).collect(java.util.stream.Collectors.toSet())); + assertEquals(3, requests.size(), "unconditional edges must bypass onRoute"); + assertEquals(4, routeEvents.size()); + assertTrue( + routeEvents.stream() + .anyMatch( + event -> + event.get("next").equals("always") + && event.get("conditional").equals(false) + && event.get("allowed").equals(true))); + assertTrue( + routeEvents.stream() + .anyMatch( + event -> + event.get("next").equals("denied") + && event.get("conditional").equals(true) + && event.get("allowed").equals(false) + && event.get("reason").equals("not applicable"))); + assertEquals( + Set.of("A", "B", "C", "D"), + result.getHistory().stream() + .map(item -> String.valueOf(item.get("agent"))) + .collect(java.util.stream.Collectors.toSet())); + } + + @Test + void whitespaceOnlyConditionIsUnconditional() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next(List.of(jump("next", " \t"))) + .build(); + WorkflowStep next = + WorkflowStep.builder().name("next").subtasks(List.of(task("B", "next"))).build(); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .build(); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().name("blank-condition").steps(List.of(start, next)).build(), + callbacks, + new StubWorkflowEngineClient("A", "B"), + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertTrue(result.isSuccess()); + assertEquals(2, result.getHistory().size()); + } + + @Test + void conditionalEdgesAreEvaluatedConcurrently() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next(List.of(jump("left", "left condition"), jump("right", "right condition"))) + .build(); + WorkflowStep left = + WorkflowStep.builder().name("left").subtasks(List.of(task("B", "left"))).build(); + WorkflowStep right = + WorkflowStep.builder().name("right").subtasks(List.of(task("C", "right"))).build(); + ExecutorService routeExecutor = Executors.newFixedThreadPool(2); + CyclicBarrier routeBarrier = new CyclicBarrier(2); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute( + request -> + CompletableFuture.supplyAsync( + () -> { + awaitBarrier(routeBarrier); + return RouteDecision.allow("matched"); + }, + routeExecutor)) + .build(); + + try { + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder() + .name("parallel-routes") + .steps(List.of(start, left, right)) + .build(), + callbacks, + new StubWorkflowEngineClient("A", "B", "C"), + recordingCallback(), + "", + "zh") + .run() + .join(); + assertTrue(result.isSuccess()); + assertEquals(3, result.getHistory().size()); + } finally { + routeExecutor.shutdownNow(); + } + } + + @Test + void routeFailurePreventsUnconditionalAndAllowedTargetsFromStarting() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next( + List.of( + jump("always", ""), + jump("failed", "cannot evaluate"), + jump("allowed", "matched"))) + .build(); + WorkflowStep always = + WorkflowStep.builder().name("always").subtasks(List.of(task("B", "always"))).build(); + WorkflowStep failed = + WorkflowStep.builder().name("failed").subtasks(List.of(task("C", "failed"))).build(); + WorkflowStep allowed = + WorkflowStep.builder().name("allowed").subtasks(List.of(task("D", "allowed"))).build(); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute( + request -> + request.nextStep().equals("failed") + ? CompletableFuture.failedFuture( + new IllegalStateException("route evaluation unavailable")) + : CompletableFuture.completedFuture(RouteDecision.allow())) + .build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A", "B", "C", "D"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder() + .name("atomic-routes") + .steps(List.of(start, always, failed, allowed)) + .build(), + callbacks, + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("start -> failed")); + assertTrue(result.getError().contains("route evaluation unavailable")); + assertEquals( + List.of("A"), client.getSentMessages().stream().map(message -> message.agentName).toList()); + } + + @Test + void conditionalRouteMayDenyEveryEdge() { WorkflowStep s1 = WorkflowStep.builder() .name("s1") @@ -338,23 +565,13 @@ public CompletableFuture onTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("nonexistent").reason("bad").build()); + return CompletableFuture.completedFuture(RouteDecision.deny("condition not met")); } }; WorkflowExecutor exec = new WorkflowExecutor(wf, cp, stub, recordingCallback(), "", "zh"); ExecutionResult result = exec.run().join(); - // s1 executes, then an invalid route is reported as a workflow error (s2 never runs). - assertFalse(result.isSuccess()); - assertTrue(result.getError().contains("allowed next steps")); + assertTrue(result.isSuccess()); assertEquals(1, stub.getSentCount()); } @@ -395,6 +612,154 @@ public CompletableFuture onRoute(RouteRequest routeRequest) { assertFalse(result.isSuccess()); assertTrue(result.getError().contains("onRoute returned null decision")); + assertTrue(result.getError().contains("s1 -> s2")); + } + + @Test + void onRouteNullFutureFailsAfterInvokingEveryConditionalEdge() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next(List.of(jump("first", "first condition"), jump("second", "second condition"))) + .build(); + WorkflowStep first = + WorkflowStep.builder().name("first").subtasks(List.of(task("B", "first"))).build(); + WorkflowStep second = + WorkflowStep.builder().name("second").subtasks(List.of(task("C", "second"))).build(); + List evaluated = Collections.synchronizedList(new ArrayList<>()); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute( + request -> { + evaluated.add(request.nextStep()); + return request.nextStep().equals("first") + ? null + : CompletableFuture.completedFuture(RouteDecision.allow()); + }) + .build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A", "B", "C"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder() + .name("null-route-future") + .steps(List.of(start, first, second)) + .build(), + callbacks, + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertEquals(List.of("first", "second"), evaluated); + assertTrue(result.getError().contains("start -> first")); + assertTrue(result.getError().contains("onRoute returned null future")); + assertEquals( + result.getError().indexOf("onRoute failed for edge"), + result.getError().lastIndexOf("onRoute failed for edge")); + assertEquals(List.of("A"), client.getSentMessages().stream().map(m -> m.agentName).toList()); + } + + @Test + void synchronousRouteExceptionFailsAfterInvokingEveryConditionalEdge() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next(List.of(jump("first", "first condition"), jump("second", "second condition"))) + .build(); + WorkflowStep first = + WorkflowStep.builder().name("first").subtasks(List.of(task("B", "first"))).build(); + WorkflowStep second = + WorkflowStep.builder().name("second").subtasks(List.of(task("C", "second"))).build(); + List evaluated = Collections.synchronizedList(new ArrayList<>()); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute( + request -> { + evaluated.add(request.nextStep()); + if (request.nextStep().equals("first")) { + throw new IllegalArgumentException("invalid routing data"); + } + return CompletableFuture.completedFuture(RouteDecision.allow()); + }) + .build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A", "B", "C"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder() + .name("synchronous-route-failure") + .steps(List.of(start, first, second)) + .build(), + callbacks, + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertEquals(List.of("first", "second"), evaluated); + assertTrue(result.getError().contains("start -> first")); + assertTrue(result.getError().contains("invalid routing data")); + assertEquals(List.of("A"), client.getSentMessages().stream().map(m -> m.agentName).toList()); + } + + @Test + void onRouteTimeoutIdentifiesTheConditionalEdge() throws Exception { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "start"))) + .next(List.of(jump("next", "slow condition"))) + .build(); + WorkflowStep next = + WorkflowStep.builder().name("next").subtasks(List.of(task("B", "next"))).build(); + ControlPoint callbacks = + ControlPoint.builder() + .onTask( + request -> + CompletableFuture.completedFuture( + MessageContent.text(request.getInstruction()))) + .onRoute(request -> new CompletableFuture<>()) + .build(); + StubWorkflowEngineClient client = + new StubWorkflowEngineClient("A", "B") { + @Override + public long callbackTimeoutSeconds() { + return 1; + } + }; + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().name("route-timeout").steps(List.of(start, next)).build(), + callbacks, + client, + recordingCallback(), + "", + "zh") + .run() + .get(3, TimeUnit.SECONDS); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("start -> next")); + assertTrue(result.getError().contains("TimeoutException")); + assertEquals(List.of("A"), client.getSentMessages().stream().map(m -> m.agentName).toList()); } @Test @@ -423,15 +788,7 @@ public CompletableFuture onTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; WorkflowExecutor exec = new WorkflowExecutor(wf, cp, stub, recordingCallback(), "", "zh"); @@ -492,15 +849,7 @@ public CompletableFuture onTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; WorkflowExecutor exec = new WorkflowExecutor(wf, cp, stub, recordingCallback(), "", "zh"); @@ -561,15 +910,7 @@ public CompletableFuture onTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; WorkflowExecutor exec = @@ -618,6 +959,86 @@ void missingStaticRouteTargetFailsBeforeExecution() { assertEquals(0, stub.getSentCount()); } + @Test + void duplicateOutgoingTargetFailsBeforeExecution() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "run"))) + .next(List.of(jump("next", ""), jump("next", "condition"))) + .build(); + WorkflowStep next = + WorkflowStep.builder().name("next").subtasks(List.of(task("B", "next"))).build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A", "B"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().name("duplicate-route").steps(List.of(start, next)).build(), + autoCp(), + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("duplicate outgoing target next")); + assertEquals(0, client.getSentCount()); + } + + @Test + void nullOutgoingEdgeFailsBeforeExecution() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "run"))) + .next(Collections.singletonList(null)) + .build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().name("null-edge").steps(List.of(start)).build(), + autoCp(), + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("null outgoing edge")); + assertEquals(0, client.getSentCount()); + } + + @Test + void blankOutgoingTargetFailsBeforeExecution() { + WorkflowStep start = + WorkflowStep.builder() + .name("start") + .subtasks(List.of(task("A", "run"))) + .next(List.of(jump(" ", ""))) + .build(); + StubWorkflowEngineClient client = new StubWorkflowEngineClient("A"); + + ExecutionResult result = + new WorkflowExecutor( + Workflow.builder().name("blank-target").steps(List.of(start)).build(), + autoCp(), + client, + recordingCallback(), + "", + "zh") + .run() + .join(); + + assertFalse(result.isSuccess()); + assertTrue(result.getError().contains("blank target")); + assertEquals(0, client.getSentCount()); + } + @Test void cyclicWorkflowFailsBeforeExecution() { WorkflowStep first = @@ -701,15 +1122,10 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("right").reason("test").build()); + routeRequest.nextStep().equals("right") + ? RouteDecision.allow("test") + : RouteDecision.deny("test")); } }; StubWorkflowEngineClient stub = new StubWorkflowEngineClient("A", "B", "C", "M"); @@ -775,15 +1191,7 @@ public CompletableFuture onSelfTask(TaskRequest req) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String s = routeRequest.stepName(); - Map r = java.util.Map.of(); - List c = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(c.get(0).getStep()).reason("first").build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; WorkflowExecutor exec = new WorkflowExecutor(wf, cp, stub, recordingCallback(), "intent", "zh"); @@ -828,15 +1236,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("end").build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; @@ -884,15 +1284,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("end").build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; ExecutionResult success = @@ -917,15 +1309,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(option -> new JumpCondition(option.nextStep(), option.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep("end").build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; ExecutionResult failure = diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/model/RouteModelTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/RouteModelTest.java new file mode 100644 index 00000000..a120060e --- /dev/null +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/model/RouteModelTest.java @@ -0,0 +1,65 @@ +/* + * Copyright (c) 2026 Huawei Technologies Co., Ltd. + * All Rights Reserved. + * + * SPDX-License-Identifier: Apache-2.0 + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may + * not use this file except in compliance with the License. You may obtain + * a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package dev.openan.workflow.engine.model; + +import static org.junit.jupiter.api.Assertions.*; + +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; + +class RouteModelTest { + + @Test + void decisionFactoriesExposeAllowedStateAndReason() { + assertEquals(new RouteDecision(true, ""), RouteDecision.allow()); + assertEquals(new RouteDecision(true, "matched"), RouteDecision.allow("matched")); + assertEquals(new RouteDecision(false, ""), RouteDecision.deny()); + assertEquals(new RouteDecision(false, "blocked"), RouteDecision.deny("blocked")); + assertEquals("", new RouteDecision(true, null).reason()); + } + + @Test + void requestRepresentsOneConditionalEdgeAndCopiesResults() { + List results = new ArrayList<>(); + RouteRequest request = + new RouteRequest( + "execution", "source", "target", "alarm exists", WorkflowInput.empty(), results); + results.add(null); + + assertEquals("target", request.nextStep()); + assertEquals("alarm exists", request.condition()); + assertEquals(List.of(), request.currentResults()); + } + + @Test + void requestRejectsBlankConditionalEdgeFields() { + assertThrows( + IllegalArgumentException.class, + () -> + new RouteRequest( + "execution", "source", "target", " ", WorkflowInput.empty(), List.of())); + assertThrows( + IllegalArgumentException.class, + () -> + new RouteRequest( + "execution", "source", " ", "condition", WorkflowInput.empty(), List.of())); + } +} diff --git a/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java b/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java index 35884167..9a017cc9 100644 --- a/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java +++ b/workflow-engine/src/test/java/dev/openan/workflow/engine/runner/ExecutePsopTest.java @@ -19,16 +19,28 @@ package dev.openan.workflow.engine.runner; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import dev.openan.workflow.engine.StubWorkflowEngineClient; import dev.openan.workflow.engine.control.ControlPoint; import dev.openan.workflow.engine.control.EventCallback; import dev.openan.workflow.engine.control.EventType; -import dev.openan.workflow.engine.model.*; +import dev.openan.workflow.engine.model.ExecutionResult; +import dev.openan.workflow.engine.model.JumpCondition; import dev.openan.workflow.engine.model.MessageContent; +import dev.openan.workflow.engine.model.RouteDecision; import dev.openan.workflow.engine.model.RouteRequest; -import java.util.*; +import dev.openan.workflow.engine.model.Task; +import dev.openan.workflow.engine.model.TaskRequest; +import dev.openan.workflow.engine.model.Workflow; +import dev.openan.workflow.engine.model.WorkflowStep; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; @@ -49,22 +61,33 @@ void cancellationStopsLateTaskPreparationAndClosesLifecycleOnce() throws Excepti var finished = new java.util.concurrent.CountDownLatch(1); var closes = new AtomicInteger(); var stub = new StubWorkflowEngineClient("A", "B"); - var execution = ExecutePsop.builder().psop(linearWorkflow()).engineClient(stub) - .controlPoint(ControlPoint.builder().onTask(request -> { - entered.countDown(); - return prepared; - }).build()) - .eventCallback(new EventCallback() { - @Override public void onEvent(String type, Map data) { - if (EventType.CLOSE.equals(type)) closes.incrementAndGet(); - } - }) - .onFinish((result, events) -> { - assertFalse(result.isSuccess()); - assertEquals("Workflow execution cancelled", result.getError()); - finished.countDown(); - return CompletableFuture.completedFuture(null); - }).execute(); + var execution = + ExecutePsop.builder() + .psop(linearWorkflow()) + .engineClient(stub) + .controlPoint( + ControlPoint.builder() + .onTask( + request -> { + entered.countDown(); + return prepared; + }) + .build()) + .eventCallback( + new EventCallback() { + @Override + public void onEvent(String type, Map data) { + if (EventType.CLOSE.equals(type)) closes.incrementAndGet(); + } + }) + .onFinish( + (result, events) -> { + assertFalse(result.isSuccess()); + assertEquals("Workflow execution cancelled", result.getError()); + finished.countDown(); + return CompletableFuture.completedFuture(null); + }) + .execute(); assertTrue(entered.await(2, java.util.concurrent.TimeUnit.SECONDS)); assertTrue(execution.cancel(true)); prepared.complete(MessageContent.text("must not send")); @@ -78,13 +101,20 @@ void cancellationStopsLateTaskPreparationAndClosesLifecycleOnce() throws Excepti void cancellationWhileOnFinishIsPendingStillClosesExactlyOnce() { var finish = new CompletableFuture(); var closes = new AtomicInteger(); - var execution = ExecutePsop.builder().psop(linearWorkflow()) - .engineClient(new StubWorkflowEngineClient("A", "B")).controlPoint(autoCp()) - .eventCallback(new EventCallback() { - @Override public void onEvent(String type, Map data) { - if (EventType.CLOSE.equals(type)) closes.incrementAndGet(); - } - }).onFinish((result, events) -> finish).execute(); + var execution = + ExecutePsop.builder() + .psop(linearWorkflow()) + .engineClient(new StubWorkflowEngineClient("A", "B")) + .controlPoint(autoCp()) + .eventCallback( + new EventCallback() { + @Override + public void onEvent(String type, Map data) { + if (EventType.CLOSE.equals(type)) closes.incrementAndGet(); + } + }) + .onFinish((result, events) -> finish) + .execute(); assertTrue(execution.cancel(true)); assertEquals(1, closes.get()); finish.complete(null); @@ -94,16 +124,22 @@ void cancellationWhileOnFinishIsPendingStillClosesExactlyOnce() { @Test void cancelledRunnerClosesItsOwnedRuntime() { var closes = new AtomicInteger(); - var runtime = (dev.openan.workflow.engine.client.A2AJavaClientRuntime) - java.lang.reflect.Proxy.newProxyInstance(getClass().getClassLoader(), - new Class[] {dev.openan.workflow.engine.client.A2AJavaClientRuntime.class}, - (proxy, method, args) -> { - if (method.getName().equals("close")) closes.incrementAndGet(); - return null; - }); - var execution = ExecutePsop.builder().psop(linearWorkflow()).a2aClientRuntime(runtime) - .controlPoint(ControlPoint.builder().onTask(request -> new CompletableFuture<>()).build()) - .execute(); + var runtime = + (dev.openan.workflow.engine.client.A2AJavaClientRuntime) + java.lang.reflect.Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[] {dev.openan.workflow.engine.client.A2AJavaClientRuntime.class}, + (proxy, method, args) -> { + if (method.getName().equals("close")) closes.incrementAndGet(); + return null; + }); + var execution = + ExecutePsop.builder() + .psop(linearWorkflow()) + .a2aClientRuntime(runtime) + .controlPoint( + ControlPoint.builder().onTask(request -> new CompletableFuture<>()).build()) + .execute(); assertTrue(execution.cancel(true)); assertEquals(1, closes.get()); } @@ -121,15 +157,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(c -> new JumpCondition(c.nextStep(), c.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).reason("first").build()); + return CompletableFuture.completedFuture(RouteDecision.allow("test")); } }; } @@ -393,15 +421,7 @@ public CompletableFuture onTask(TaskRequest request) { @Override public CompletableFuture onRoute(RouteRequest routeRequest) { - String stepName = routeRequest.stepName(); - Map results = java.util.Map.of(); - List conditions = - routeRequest.candidates().stream() - .map(c -> new JumpCondition(c.nextStep(), c.condition())) - .toList(); - - return CompletableFuture.completedFuture( - RouteDecision.builder().nextStep(conditions.get(0).getStep()).build()); + return CompletableFuture.completedFuture(RouteDecision.allow()); } }; List events = Collections.synchronizedList(new ArrayList<>());