1. 重构了工作流执行模型，节点完成后立即触发后继节点，不再被同层慢节点阻塞等待，提升并行执行效率；

2. 实现了条件分支未选路径上的节点自动标记为 SKIPPED，前端可清晰看到分支选择结果；
3. 优化了节点失败 + abort 策略下的执行收尾，所有未启动节点统一显示为 SKIPPED 状态，便于查看工作流中断点；
4. 新增了工作流执行引擎的单元测试，覆盖数据流调度、失败策略、条件分支等核心场景；
5. 新增了工作流数据流改造实施计划文档；
6. 更新了 prompt.md 需求记录。

backend/src/main/java/com/agent/management/engine/WorkflowLevelExecutor.java

@@ -13,20 +13,35 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 import java.io.IOException;
 import java.nio.file.Path;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 /**
- * 按 DAG 拓扑层级逐层执行节点。
- * 从 WorkflowEngine.execute() 抽取，承担节点遍历、状态推进与失败策略处理。
+ * 按 DAG 拓扑执行节点（Dataflow 模型）。
+ *
+ * <p>调度规则：节点所有前驱完成后立即被调度，不等同层兄弟节点。
+ * 通过 {@link CompletableFuture#whenComplete} 回调链驱动后继节点；
+ * 不再有"层"barrier，慢节点不再阻塞快节点的下游链路。</p>
+ *
+ * <p>线程模型：</p>
+ * <ul>
+ *   <li>{@link #nodeExecutor}：节点 worker 线程池，{@link #executeOneNode} 在此执行</li>
+ *   <li>回调链（onNodeCompleted / tryScheduleSuccessors）也在 nodeExecutor 中执行</li>
+ *   <li>主线程仅在 {@code ctx.done.get(...)} 上等待</li>
+ * </ul>
  */
 @Slf4j
 @Component
@@ -37,12 +52,15 @@ public class WorkflowLevelExecutor {
     private static final ObjectMapper MAPPER = new ObjectMapper()
             .registerModule(new JavaTimeModule());
 
+    /** dataflow 调度主线程等待超时（兜底，避免死锁时永久阻塞） */
+    private static final long EXECUTION_TIMEOUT_MINUTES = 30L;
+
     private final Map<String, NodeExecutor> executorMap;
     private final SseEventBus eventBus;
     private final NodeWorkspaceBuilder workspaceBuilder;
     /**
-     * 同层节点并行执行的线程池。独立于 WorkflowEngine.executor（工作流主线程池），
-     * 避免层内并行与层间串行争用同一池导致死锁。
+     * 节点执行的线程池。独立于 WorkflowEngine.executor（工作流主线程池），
+     * 避免节点任务与外层 wrapper 争用同一池导致死锁。
      */
     private final ExecutorService nodeExecutor = Executors.newFixedThreadPool(
             Math.max(8, Runtime.getRuntime().availableProcessors() * 2),
@@ -100,7 +118,68 @@ public class WorkflowLevelExecutor {
     }
 
     /**
-     * 按层执行 DAG。
+     * Dataflow 调度的共享上下文。
+     *
+     * <p>所有并发访问的可变状态集中在此对象，便于审计与生命周期管理。
+     * dag / context / runId / runRecordId / emitter 为不可变引用，回调链中通过 ctx.dag / ctx.context 直接访问。</p>
+     */
+    private static final class DataflowCtx {
+        final DagResolver.ResolvedDag dag;
+        final WorkflowContext context;
+        final String runId;
+        final Long runRecordId;
+        final SseEmitter emitter;
+
+        /** 活跃节点集合（待执行或执行中）；非活跃节点入度归零后会被标记 SKIPPED */
+        final Set<String> activeNodes = ConcurrentHashMap.newKeySet();
+        /** 活跃边集合（条件分支按选中 sourceHandle 过滤后实际生效的边） */
+        final Set<String> activeEdges = ConcurrentHashMap.newKeySet();
+        /** 节点的剩余入度（按 source 去重，与 DagResolver 入度规则一致）；归零则可调度 */
+        final Map<String, AtomicInteger> pendingInputs = new ConcurrentHashMap<>();
+        /** nodeId → 在飞 future；abort 时用于 cancel */
+        final Map<String, CompletableFuture<NodeExecutionResult>> inFlightFutures = new ConcurrentHashMap<>();
+        /** 已完成（含 SKIPPED）的节点集合；用于去重 onNodeCompleted 与 triggerAbort 之间的双重处理 */
+        final Set<String> completedNodes = ConcurrentHashMap.newKeySet();
+
+        /** 剩余未完成节点数；归零时唤醒主线程（complete done） */
+        final AtomicInteger remainingNodes;
+        /** abort 标志（CAS 去重，保证 workflow_error 只推一次） */
+        final AtomicBoolean abortFlag = new AtomicBoolean(false);
+        /** abort 错误信息（首个失败的 abort 节点写入） */
+        final AtomicReference<String> abortMsgRef = new AtomicReference<>(null);
+        /** 全局递增的 sortOrder（按完成顺序赋值） */
+        final AtomicInteger sortOrderSeq = new AtomicInteger(0);
+
+        /** 最终输出（output 节点完成时写入）；持锁访问 */
+        final Map<String, Object> finalOutputs = new LinkedHashMap<>();
+        /** 节点执行记录；持锁访问 */
+        final List<WorkflowRunNode> nodeRecords = new ArrayList<>();
+        /** 保护 finalOutputs 与 nodeRecords 的锁 */
+        final Object recordsLock = new Object();
+
+        /** 主线程等待句柄；remainingNodes 归零时 complete */
+        final CompletableFuture<Void> done = new CompletableFuture<>();
+
+        DataflowCtx(DagResolver.ResolvedDag dag, WorkflowContext context,
+                    String runId, Long runRecordId, SseEmitter emitter) {
+            this.dag = dag;
+            this.context = context;
+            this.runId = runId;
+            this.runRecordId = runRecordId;
+            this.emitter = emitter;
+            this.remainingNodes = new AtomicInteger(dag.getNodeDataMap().size());
+        }
+
+        /** 标记一个节点完成（含 skipped）；remainingNodes 归零时唤醒主线程 */
+        void markCompleted() {
+            if (remainingNodes.decrementAndGet() == 0) {
+                done.complete(null);
+            }
+        }
+    }
+
+    /**
+     * 按 DAG 数据流驱动执行。
      *
      * @param dag         已解析的 DAG
      * @param context     工作流上下文
@@ -111,105 +190,287 @@ public class WorkflowLevelExecutor {
      */
     public ExecutionOutcome executeByLevel(DagResolver.ResolvedDag dag, WorkflowContext context,
                                            String runId, Long runRecordId, SseEmitter emitter) {
-        Set<String> activeNodes = new HashSet<>();
-        if (!dag.getLevels().isEmpty()) {
-            activeNodes.addAll(dag.getLevels().get(0));
-        }
-
-        // 运行时活跃边集合，用于条件分支的动态可达集计算
-        Set<String> activeEdges = new HashSet<>();
+        DataflowCtx ctx = new DataflowCtx(dag, context, runId, runRecordId, emitter);
 
         // 注入流式回调：executor 调用 sink.emit(...) 时实时包装为 node_stream 事件推给前端
         context.setStreamSink((nodeId, kind, content, toolName) ->
                 safeSend(emitter, WorkflowRunEvent.nodeStream(runId, nodeId, kind, content, toolName)));
 
-        Map<String, Object> finalOutputs = new LinkedHashMap<>();
-        List<WorkflowRunNode> nodeRecords = new ArrayList<>();
-        int sortOrder = 0;
+        // === 初始化入度（按 source 去重，与 DagResolver 入度计算规则保持一致） ===
+        for (Map.Entry<String, List<DagResolver.EdgeInfo>> e : dag.getIncomingEdges().entrySet()) {
+            Set<String> sources = new HashSet<>();
+            for (DagResolver.EdgeInfo edge : e.getValue()) {
+                sources.add(edge.getSource());
+            }
+            ctx.pendingInputs.put(e.getKey(), new AtomicInteger(sources.size()));
+        }
+        for (String nodeId : dag.getNodeDataMap().keySet()) {
+            ctx.pendingInputs.computeIfAbsent(nodeId, k -> new AtomicInteger(0));
+        }
+
+        // === 所有节点初始都视为活跃；条件分支未选 target 由 tryScheduleSuccessors 标记移除，
+        // === SKIPPED 节点的下游由 handleSkippedNode 传播时移除 ===
+        for (String nodeId : dag.getNodeDataMap().keySet()) {
+            ctx.activeNodes.add(nodeId);
+        }
+        // 入度为 0 的节点（DAG 起点）提交调度
+        for (Map.Entry<String, AtomicInteger> e : ctx.pendingInputs.entrySet()) {
+            if (e.getValue().get() == 0) {
+                scheduleNode(e.getKey(), ctx);
+            }
+        }
 
-        for (int levelIdx = 0; levelIdx < dag.getLevels().size(); levelIdx++) {
-            List<String> level = dag.getLevels().get(levelIdx);
-            log.debug("[WorkflowEngine] 执行第 {} 层, {} 个节点, 活跃: {}", levelIdx, level.size(),
-                    level.stream().filter(activeNodes::contains).count());
+        // === 主线程等待所有节点完成（含 abort 后的清理） ===
+        try {
+            ctx.done.get(EXECUTION_TIMEOUT_MINUTES, TimeUnit.MINUTES);
+        } catch (TimeoutException e) {
+            log.error("[WorkflowEngine] 工作流执行超时（{} 分钟），强制结束: runId={}",
+                    EXECUTION_TIMEOUT_MINUTES, runId);
+            triggerAbort(ctx, "工作流执行超时");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("[WorkflowEngine] 工作流执行被中断: runId={}", runId);
+        } catch (ExecutionException e) {
+            log.error("[WorkflowEngine] 工作流执行异常: runId={}", runId, e);
+        }
+
+        String errorMsg = ctx.abortMsgRef.get();
+        return new ExecutionOutcome(ctx.finalOutputs, ctx.nodeRecords, errorMsg != null, errorMsg);
+    }
 
-            // 1. 分离活跃与非活跃节点；非活跃节点直接标记为 SKIPPED（保持原 sortOrder 顺序）
-            List<String> activeInLevel = new ArrayList<>();
-            for (String nodeId : level) {
-                if (activeNodes.contains(nodeId)) {
-                    activeInLevel.add(nodeId);
+    /**
+     * 调度单个节点：提交到线程池执行，挂载 whenComplete 回调。
+     *
+     * <p>并发安全：</p>
+     * <ul>
+     *   <li>abort 时直接 return，不再提交新任务，未启动节点转交 handleSkippedNode 收尾</li>
+     *   <li>重复调度同一节点（理论不应发生）由 inFlightFutures 的 putIfAbsent 检测</li>
+     * </ul>
+     */
+    private void scheduleNode(String nodeId, DataflowCtx ctx) {
+        if (ctx.abortFlag.get()) {
+            // 已 abort：未启动节点不再调度，直接按 skipped 处理（仅递减 remaining，不写 record/SSE）
+            handleSkippedNode(nodeId, ctx);
+            return;
+        }
+        CompletableFuture<NodeExecutionResult> existing = ctx.inFlightFutures.get(nodeId);
+        if (existing != null) {
+            // 已在飞 / 已处理，避免重复调度
+            return;
+        }
+
+        CompletableFuture<NodeExecutionResult> future = CompletableFuture.supplyAsync(
+                () -> executeOneNode(nodeId, ctx.dag, ctx.context, ctx.runId, ctx.emitter, ctx.activeEdges),
+                nodeExecutor);
+        // putIfAbsent 防止并发场景下重复提交（如多前驱同时归零）
+        CompletableFuture<NodeExecutionResult> race = ctx.inFlightFutures.putIfAbsent(nodeId, future);
+        if (race != null) {
+            // 极端竞态：另一线程已先提交，取消当前 future 并复用既有
+            future.cancel(false);
+            return;
+        }
+        future.whenComplete((result, ex) -> onNodeCompleted(nodeId, ctx, result, ex));
+    }
+
+    /**
+     * 节点完成回调：处理结果（成功/失败/异常）、写 records、推 SSE、激活下游。
+     *
+     * <p>无论 abort 与否，都会执行 markCompleted；abort 时仅清理 inFlightFutures，
+     * 不写 records / 不推 SSE / 不激活下游（避免与已发的 workflow_error 乱序）。</p>
+     */
+    private void onNodeCompleted(String nodeId, DataflowCtx ctx,
+                                 NodeExecutionResult result, Throwable ex) {
+        ctx.inFlightFutures.remove(nodeId);
+        // 去重：同一节点可能因 cancel + 正常完成进入两次回调
+        if (!ctx.completedNodes.add(nodeId)) {
+            return;
+        }
+
+        // 异常降级为 failed 结果
+        if (ex != null) {
+            result = NodeExecutionResult.failed(nodeId, "节点并行执行异常: " + ex.getMessage());
+        } else if (result == null) {
+            result = NodeExecutionResult.failed(nodeId, "节点执行返回 null");
+        }
+
+        // abort 后只清理计数，不处理副作用
+        if (ctx.abortFlag.get()) {
+            ctx.markCompleted();
+            return;
+        }
+
+        String nodeType = ctx.dag.getNodeTypeMap().get(nodeId);
+        JsonNode nodeData = ctx.dag.getNodeDataMap().get(nodeId);
+        String label = nodeData.path("label").asText(nodeId);
+
+        // 持锁处理 records / finalOutputs / sortOrder（dataflow 下多 worker 并发进入）
+        boolean shouldAbort = false;
+        String abortMsg = null;
+        synchronized (ctx.recordsLock) {
+            int sortOrder = ctx.sortOrderSeq.getAndIncrement();
+            ctx.nodeRecords.add(buildNodeRecord(ctx.runRecordId, nodeId, nodeType, label,
+                    result.getStatus().name(), result.getOutput(), result.getError(),
+                    result.getSelectedBranch(), result.getLogs(), result.getContextSnapshot(), sortOrder));
+
+            if (result.getStatus() == NodeExecutionResult.Status.FAILED) {
+                String failStrategy = nodeData.path("failStrategy").asText("abort");
+                if ("skip".equals(failStrategy)) {
+                    log.info("[WorkflowEngine] 节点 {} 执行失败，策略为跳过，继续工作流", nodeId);
                 } else {
-                    String nodeType = dag.getNodeTypeMap().get(nodeId);
-                    JsonNode nodeData = dag.getNodeDataMap().get(nodeId);
-                    String label = nodeData.path("label").asText(nodeId);
-                    safeSend(emitter, WorkflowRunEvent.nodeResult(runId, NodeExecutionResult.skipped(nodeId)));
-                    nodeRecords.add(buildNodeRecord(runRecordId, nodeId, nodeType, label,
-                            "SKIPPED", null, null, null, null, null, sortOrder++));
+                    shouldAbort = true;
+                    abortMsg = "节点 " + nodeId + " 执行失败: " + result.getError();
                 }
             }
 
-            if (activeInLevel.isEmpty()) continue;
+            if ("output".equals(nodeType) && result.getOutput() != null) {
+                // result.getOutput() 已是 envelope.toMap()，整体作为工作流最终输出
+                ctx.finalOutputs.clear();
+                ctx.finalOutputs.putAll(result.getOutput());
+            }
+        }
+
+        // 推送 nodeResult（在 recordsLock 之外，避免与 safeSend 形成嵌套锁）
+        safeSend(ctx.emitter, WorkflowRunEvent.nodeResult(ctx.runId, result));
+
+        // 失败 + abort：跳过下游调度，直接 triggerAbort（未启动节点由 triggerAbort 内 handleSkippedNode 收尾）
+        if (shouldAbort) {
+            triggerAbort(ctx, abortMsg);
+            ctx.markCompleted();
+            return;
+        }
+
+        // 激活下游（含条件分支入度补偿）
+        tryScheduleSuccessors(nodeId, nodeType, result, ctx);
 
-            // 2. 同层活跃节点并行执行：每个节点提交到 nodeExecutor 线程池
-            List<CompletableFuture<NodeExecutionResult>> futures = new ArrayList<>(activeInLevel.size());
-            Set<String> capturedActiveEdges = activeEdges;
-            for (String nodeId : activeInLevel) {
-                final String finalNodeId = nodeId;
-                futures.add(CompletableFuture.supplyAsync(
-                        () -> executeOneNode(finalNodeId, dag, context, runId, emitter, capturedActiveEdges),
-                        nodeExecutor));
+        ctx.markCompleted();
+    }
+
+    /**
+     * 激活后继节点：递减 target 入度，归零则调度。
+     *
+     * <p>关键规则（与 DagResolver 入度去重保持一致）：</p>
+     * <ul>
+     *   <li>每个 source 完成时，对其所有 target 各递减 1 次（即使同 source 多条边到同 target）</li>
+     *   <li>条件分支节点：未选中分支的 target 仍要递减入度，但 activeEdges 不加</li>
+     *   <li>顺序：先 add activeEdges 再 decrement 入度，保证新 worker 启动时能看到活跃边（happens-before）</li>
+     * </ul>
+     */
+    private void tryScheduleSuccessors(String nodeId, String nodeType, NodeExecutionResult result,
+                                       DataflowCtx ctx) {
+        List<DagResolver.EdgeInfo> outEdges = ctx.dag.getOutgoingEdges().get(nodeId);
+        if (outEdges == null || outEdges.isEmpty()) return;
+
+        boolean isConditionBranch = "condition".equals(nodeType) && result.getSelectedBranch() != null;
+        // 同一 target 在本节点完成时只递减一次（无论几条边、几个分支）
+        Set<String> decreased = new HashSet<>();
+
+        // 条件分支：先把未选 target 从 activeNodes 移除，使其入度归零时走 SKIPPED 路径而非调度
+        if (isConditionBranch) {
+            for (DagResolver.EdgeInfo edge : outEdges) {
+                if (!result.getSelectedBranch().equals(edge.getSourceHandle())) {
+                    ctx.activeNodes.remove(edge.getTarget());
+                }
+            }
+        }
+
+        for (DagResolver.EdgeInfo edge : outEdges) {
+            boolean edgeActive = !isConditionBranch
+                    || result.getSelectedBranch().equals(edge.getSourceHandle());
+            if (edgeActive) {
+                // 先加活跃边，确保下游 worker 构建工作区时能看到（happens-before）
+                ctx.activeEdges.add(WorkflowScopeResolver.activeEdgeKey(
+                        edge.getSource(), edge.getSourceHandle(), edge.getTarget()));
             }
-            // 等待当前层所有并行节点完成（任一节点抛出的异常都被封装为 failed 结果，不会从 join 传播）
-            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
-
-            // 3. 串行处理结果：按提交顺序写 nodeRecords / sortOrder，按 failStrategy 决定是否 abort
-            //    结果处理阶段是单线程，无需同步 nodeRecords / finalOutputs / sortOrder
-            String abortMsg = null;
-            for (int i = 0; i < activeInLevel.size(); i++) {
-                String nodeId = activeInLevel.get(i);
-                NodeExecutionResult result = joinResult(futures.get(i), nodeId);
-                String nodeType = dag.getNodeTypeMap().get(nodeId);
-                JsonNode nodeData = dag.getNodeDataMap().get(nodeId);
-                String label = nodeData.path("label").asText(nodeId);
-
-                safeSend(emitter, WorkflowRunEvent.nodeResult(runId, result));
-                nodeRecords.add(buildNodeRecord(runRecordId, nodeId, nodeType, label,
-                        result.getStatus().name(), result.getOutput(), result.getError(),
-                        result.getSelectedBranch(), result.getLogs(), result.getContextSnapshot(), sortOrder++));
-
-                if (result.getStatus() == NodeExecutionResult.Status.FAILED) {
-                    String failStrategy = nodeData.path("failStrategy").asText("abort");
-                    if ("skip".equals(failStrategy)) {
-                        log.info("[WorkflowEngine] 节点 {} 执行失败，策略为跳过，继续工作流", nodeId);
-                    } else if (abortMsg == null) {
-                        // 同层多个失败时，仅记录第一个 abort 原因；继续处理剩余结果以保证 nodeRecords 完整
-                        abortMsg = "节点 " + nodeId + " 执行失败: " + result.getError();
+            if (decreased.add(edge.getTarget())) {
+                AtomicInteger pending = ctx.pendingInputs.get(edge.getTarget());
+                if (pending != null && pending.decrementAndGet() == 0) {
+                    // 入度归零：决定调度或 SKIPPED
+                    if (ctx.activeNodes.contains(edge.getTarget())) {
+                        scheduleNode(edge.getTarget(), ctx);
+                    } else {
+                        handleSkippedNode(edge.getTarget(), ctx);
                     }
                 }
+            }
+        }
+    }
 
-                // 激活下游节点并记录活跃边
-                activateDownstream(nodeId, nodeType, result, dag, activeNodes, activeEdges);
+    /**
+     * 处理 SKIPPED 节点：推送 skipped SSE + 写 record + 继续传播下游入度。
+     *
+     * <p>触发场景：</p>
+     * <ol>
+     *   <li>条件分支未选中路径上的节点（前驱完成时发现 target 不在 activeNodes 中）</li>
+     *   <li>abort 后未启动的节点（scheduleNode 入口检测 abortFlag 后直接走此分支）</li>
+     * </ol>
+     */
+    private void handleSkippedNode(String nodeId, DataflowCtx ctx) {
+        if (!ctx.completedNodes.add(nodeId)) {
+            return;  // 已处理（如 abort 时已遍历过）
+        }
+        // 标记本节点为非活跃，确保其下游入度归零时也走 SKIPPED 而不是误调度
+        ctx.activeNodes.remove(nodeId);
+        String nodeType = ctx.dag.getNodeTypeMap().get(nodeId);
+        JsonNode nodeData = ctx.dag.getNodeDataMap().get(nodeId);
+        String label = nodeData.path("label").asText(nodeId);
+        NodeExecutionResult skipped = NodeExecutionResult.skipped(nodeId);
+        // 始终写 record（abort 时也写，让前端能看到所有节点的最终状态）；
+        // 但 abort 时不推 SSE（避免与已发的 workflow_error 乱序）
+        synchronized (ctx.recordsLock) {
+            int sortOrder = ctx.sortOrderSeq.getAndIncrement();
+            ctx.nodeRecords.add(buildNodeRecord(ctx.runRecordId, nodeId, nodeType, label,
+                    "SKIPPED", null, null, null, null, null, sortOrder));
+        }
+        if (!ctx.abortFlag.get()) {
+            safeSend(ctx.emitter, WorkflowRunEvent.nodeResult(ctx.runId, skipped));
+        }
 
-                if ("output".equals(nodeType) && result.getOutput() != null) {
-                    // result.getOutput() 已是 envelope.toMap()，整体作为工作流最终输出
-                    // （含 status / message / data 三个壳字段）
-                    finalOutputs.clear();
-                    finalOutputs.putAll(result.getOutput());
+        // 传播下游：本节点 SKIPPED 等价于"前驱已处理"，对下游入度递减；
+        // 同时把下游 target 也标记为非活跃（即使其他前驱正常完成，target 也应 SKIPPED）
+        List<DagResolver.EdgeInfo> outEdges = ctx.dag.getOutgoingEdges().get(nodeId);
+        if (outEdges != null) {
+            Set<String> decreased = new HashSet<>();
+            for (DagResolver.EdgeInfo edge : outEdges) {
+                ctx.activeNodes.remove(edge.getTarget());
+                if (decreased.add(edge.getTarget())) {
+                    AtomicInteger pending = ctx.pendingInputs.get(edge.getTarget());
+                    if (pending != null && pending.decrementAndGet() == 0) {
+                        handleSkippedNode(edge.getTarget(), ctx);
+                    }
                 }
             }
+        }
+        ctx.markCompleted();
+    }
 
-            if (abortMsg != null) {
-                safeSend(emitter, WorkflowRunEvent.workflowError(runId, abortMsg));
-                return new ExecutionOutcome(finalOutputs, nodeRecords, true, abortMsg);
-            }
+    /**
+     * 触发 abort：CAS 去重 + 推送 workflow_error（仅一次）+ 取消在飞 future + 收尾未启动节点。
+     */
+    private void triggerAbort(DataflowCtx ctx, String msg) {
+        if (!ctx.abortFlag.compareAndSet(false, true)) {
+            return;  // 已 abort，仅首个失败节点推 workflow_error
         }
+        ctx.abortMsgRef.set(msg);
+        safeSend(ctx.emitter, WorkflowRunEvent.workflowError(ctx.runId, msg));
 
-        return new ExecutionOutcome(finalOutputs, nodeRecords, false, null);
+        // 取消所有在飞 future（cancel 只能中断响应中断的任务，对已返回的不影响）
+        for (Map.Entry<String, CompletableFuture<NodeExecutionResult>> e : ctx.inFlightFutures.entrySet()) {
+            e.getValue().cancel(true);
+        }
+
+        // 未启动的节点（pendingInputs 未归零、不在 inFlightFutures、不在 completedNodes）按 SKIPPED 收尾
+        // 保证 remainingNodes 最终归零，主线程能退出
+        for (String nodeId : ctx.pendingInputs.keySet()) {
+            if (!ctx.completedNodes.contains(nodeId) && !ctx.inFlightFutures.containsKey(nodeId)) {
+                handleSkippedNode(nodeId, ctx);
+            }
+        }
     }
 
     /**
      * 执行单个节点（线程池 worker 中调用）。
      * 把节点类型查找、前置条件校验、executor.execute、上下文写入等放在同一个并行任务里。
-     * 异常一律封装为 failed 结果返回，不向上抛（避免中断 CompletableFuture.allOf）。
+     * 异常一律封装为 failed 结果返回，不向上抛（避免中断 CompletableFuture 链）。
      */
     private NodeExecutionResult executeOneNode(String nodeId, DagResolver.ResolvedDag dag,
                                                WorkflowContext context, String runId,
@@ -274,38 +535,6 @@ public class WorkflowLevelExecutor {
         }
     }
 
-    /**
-     * 从 CompletableFuture 取出结果；future 内部异常一律降级为 failed。
-     */
-    private static NodeExecutionResult joinResult(CompletableFuture<NodeExecutionResult> future, String nodeId) {
-        try {
-            return future.join();
-        } catch (Exception e) {
-            return NodeExecutionResult.failed(nodeId, "节点并行执行异常: " + e.getMessage());
-        }
-    }
-
-    /**
-     * 根据当前节点执行结果激活下游节点；条件分支按选中 sourceHandle 过滤，
-     * 同时把实际激活的边加入 activeEdges 用于下游节点的动态可达集计算。
-     */
-    private void activateDownstream(String nodeId, String nodeType, NodeExecutionResult result,
-                                    DagResolver.ResolvedDag dag, Set<String> activeNodes,
-                                    Set<String> activeEdges) {
-        List<DagResolver.EdgeInfo> outEdges = dag.getOutgoingEdges().get(nodeId);
-        if (outEdges == null) return;
-
-        boolean isConditionBranch = "condition".equals(nodeType) && result.getSelectedBranch() != null;
-        for (DagResolver.EdgeInfo edge : outEdges) {
-            if (isConditionBranch && !result.getSelectedBranch().equals(edge.getSourceHandle())) {
-                continue;
-            }
-            activeNodes.add(edge.getTarget());
-            activeEdges.add(WorkflowScopeResolver.activeEdgeKey(
-                    edge.getSource(), edge.getSourceHandle(), edge.getTarget()));
-        }
-    }
-
     private WorkflowRunNode buildNodeRecord(Long runId, String nodeId, String nodeType, String label,
                                             String status, Map<String, Object> output, String error,
                                             String selectedBranch, List<ExecutionLog> logs,
@@ -432,7 +661,7 @@ public class WorkflowLevelExecutor {
 
     /**
      * 安全发送 SSE 事件
-     * <p>加 synchronized：同层多节点并行执行时，多个 worker 线程会并发调用 safeSend
+     * <p>加 synchronized：dataflow 下多 worker 并发调用 safeSend
      * （nodeRunning / nodeStream / nodeResult），而 SseEmitter.send 非线程安全。</p>
      */
     private synchronized void safeSend(SseEmitter emitter, WorkflowRunEvent event) {

backend/src/test/java/com/agent/management/engine/WorkflowLevelExecutorTest.java

@@ -0,0 +1,367 @@
+package com.agent.management.engine;
+
+import com.agent.management.external.SseEventBus;
+import com.agent.management.model.entity.WorkflowRunNode;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@link WorkflowLevelExecutor} dataflow 调度模型的核心场景测试。
+ *
+ * <p>覆盖 5 个关键场景：</p>
+ * <ol>
+ *   <li>菱形 DAG：dataflow 下 D 在 C 完成后立即调度，不等慢节点 B</li>
+ *   <li>abort 策略：失败节点阻止下游调度</li>
+ *   <li>skip 策略：失败节点仍允许下游执行</li>
+ *   <li>条件分支未选中：未选分支目标节点标记 SKIPPED 并传播</li>
+ *   <li>merge 节点入度归零：等所有前驱完成才调度</li>
+ * </ol>
+ *
+ * <p>通过 mock NodeExecutor 按 nodeId 返回不同结果，真实 DagResolver / WorkflowContext，
+ * 验证 sortOrder、status、是否调度等行为。</p>
+ */
+class WorkflowLevelExecutorTest {
+
+    private SseEventBus eventBus;
+    private NodeWorkspaceBuilder workspaceBuilder;
+    private SseEmitter emitter;
+    private Map<String, NodeExecutor> executorsMap;
+    private Map<String, Integer> sortOrderCapture;
+    private Map<String, String> statusCapture;
+
+    @BeforeEach
+    void setUp() {
+        eventBus = mock(SseEventBus.class);
+        workspaceBuilder = mock(NodeWorkspaceBuilder.class);
+        when(workspaceBuilder.build(anyString(), any(), any(), any()))
+                .thenReturn(new NodeWorkspace());
+        emitter = mock(SseEmitter.class);
+        executorsMap = new HashMap<>();
+        sortOrderCapture = new ConcurrentHashMap<>();
+        statusCapture = new ConcurrentHashMap<>();
+    }
+
+    /**
+     * 辅助：构造 mock NodeExecutor，按 nodeId 返回不同结果。
+     */
+    private NodeExecutor mockExecutor(String type, Function<String, NodeExecutionResult> behavior) {
+        NodeExecutor exec = mock(NodeExecutor.class);
+        when(exec.getType()).thenReturn(type);
+        when(exec.execute(anyString(), any(), any(), any())).thenAnswer(inv -> {
+            String nodeId = inv.getArgument(0);
+            return behavior.apply(nodeId);
+        });
+        return exec;
+    }
+
+    /**
+     * 辅助：执行 DAG 并捕获每个节点的 sortOrder / status。
+     */
+    private WorkflowLevelExecutor.ExecutionOutcome runDag(String graphData) {
+        DagResolver.ResolvedDag dag = DagResolver.resolve(graphData);
+        WorkflowContext context = new WorkflowContext(1L, "run-1", null, new HashMap<>());
+        WorkflowLevelExecutor executor = new WorkflowLevelExecutor(
+                new LinkedHashMap<>(executorsMap), eventBus, workspaceBuilder);
+        WorkflowLevelExecutor.ExecutionOutcome outcome = executor.executeByLevel(
+                dag, context, "run-1", 1L, emitter);
+
+        for (WorkflowRunNode node : outcome.nodeRecords) {
+            sortOrderCapture.put(node.getNodeId(), node.getSortOrder());
+            statusCapture.put(node.getNodeId(), node.getStatus());
+        }
+        return outcome;
+    }
+
+    /**
+     * 辅助：构造 envelope 格式的输出（status=200 + data）。
+     */
+    private static Map<String, Object> envelope(String outputValue) {
+        Map<String, Object> data = new LinkedHashMap<>();
+        data.put("output", outputValue);
+        Map<String, Object> env = new LinkedHashMap<>();
+        env.put("status", 200);
+        env.put("message", "ok");
+        env.put("data", data);
+        return env;
+    }
+
+    // ============================ 测试用例 ============================
+
+    /**
+     * 场景 1：fork chain（A→B→D 和 A→C→E），B 固定 300ms，C 立即完成。
+     *
+     * <p>dataflow 下 C 完成立即触发 E（不等 B 完成），因此 E.sortOrder 应小于 B.sortOrder。
+     * 若为 BSP 模型，B、C 同层，E 必须等 B 完成才能开始，E.sortOrder > B.sortOrder。</p>
+     */
+    @Test
+    void forkChain_fastBranchDownstreamTriggersImmediately() throws Exception {
+        String graphData = "{"
+                + "\"nodes\":["
+                + "  {\"id\":\"A\",\"type\":\"llm\",\"data\":{\"label\":\"A\"}},"
+                + "  {\"id\":\"B\",\"type\":\"llm\",\"data\":{\"label\":\"B\"}},"
+                + "  {\"id\":\"C\",\"type\":\"llm\",\"data\":{\"label\":\"C\"}},"
+                + "  {\"id\":\"D\",\"type\":\"llm\",\"data\":{\"label\":\"D\"}},"
+                + "  {\"id\":\"E\",\"type\":\"llm\",\"data\":{\"label\":\"E\"}}"
+                + "],\"edges\":["
+                + "  {\"source\":\"A\",\"target\":\"B\"},"
+                + "  {\"source\":\"A\",\"target\":\"C\"},"
+                + "  {\"source\":\"B\",\"target\":\"D\"},"
+                + "  {\"source\":\"C\",\"target\":\"E\"}"
+                + "]}";
+
+        NodeExecutor llm = mockExecutor("llm", nodeId -> {
+            try {
+                switch (nodeId) {
+                    case "A":
+                        return NodeExecutionResult.success("A", envelope("A-done"));
+                    case "B":
+                        // B 固定 300ms 延迟（模拟"慢节点"）
+                        Thread.sleep(300);
+                        return NodeExecutionResult.success("B", envelope("B-done"));
+                    case "C":
+                        return NodeExecutionResult.success("C", envelope("C-done"));
+                    case "D":
+                        return NodeExecutionResult.success("D", envelope("D-done"));
+                    case "E":
+                        return NodeExecutionResult.success("E", envelope("E-done"));
+                    default:
+                        return NodeExecutionResult.failed(nodeId, "未知节点");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return NodeExecutionResult.failed(nodeId, "中断");
+            }
+        });
+        executorsMap.put("llm", llm);
+
+        WorkflowLevelExecutor.ExecutionOutcome outcome = runDag(graphData);
+
+        assertThat(outcome.failed).isFalse();
+        assertThat(statusCapture).containsEntry("A", "SUCCESS")
+                .containsEntry("B", "SUCCESS")
+                .containsEntry("C", "SUCCESS")
+                .containsEntry("D", "SUCCESS")
+                .containsEntry("E", "SUCCESS");
+        // 关键：E（C 的下游）在 B 之前完成（dataflow 下 C 完成立即触发 E）
+        assertThat(sortOrderCapture.get("E"))
+                .as("E 应在 B 之前完成（dataflow 下 C 完成立即触发 E）")
+                .isLessThan(sortOrderCapture.get("B"));
+        // A 是起点，sortOrder=0
+        assertThat(sortOrderCapture.get("A")).isEqualTo(0);
+        // 全部 5 个节点都有 sortOrder
+        assertThat(sortOrderCapture).hasSize(5);
+    }
+
+    /**
+     * 场景 2：A→B→C，A 失败 + failStrategy=abort。
+     *
+     * <p>预期：B 和 C 不被调度（executor.execute 不被调用），workflow 标记为 failed，
+     * finalOutputs 为空。</p>
+     */
+    @Test
+    void failedNodeWithAbort_blocksDownstream() {
+        String graphData = "{"
+                + "\"nodes\":["
+                + "  {\"id\":\"A\",\"type\":\"llm\",\"data\":{\"label\":\"A\",\"failStrategy\":\"abort\"}},"
+                + "  {\"id\":\"B\",\"type\":\"llm\",\"data\":{\"label\":\"B\"}},"
+                + "  {\"id\":\"C\",\"type\":\"llm\",\"data\":{\"label\":\"C\"}}"
+                + "],\"edges\":["
+                + "  {\"source\":\"A\",\"target\":\"B\"},"
+                + "  {\"source\":\"B\",\"target\":\"C\"}"
+                + "]}";
+
+        AtomicInteger bExecutions = new AtomicInteger();
+        AtomicInteger cExecutions = new AtomicInteger();
+        NodeExecutor llm = mockExecutor("llm", nodeId -> {
+            switch (nodeId) {
+                case "A":
+                    return NodeExecutionResult.failed("A", "故意失败");
+                case "B":
+                    bExecutions.incrementAndGet();
+                    return NodeExecutionResult.success("B", envelope("B-done"));
+                case "C":
+                    cExecutions.incrementAndGet();
+                    return NodeExecutionResult.success("C", envelope("C-done"));
+                default:
+                    return NodeExecutionResult.failed(nodeId, "未知");
+            }
+        });
+        executorsMap.put("llm", llm);
+
+        WorkflowLevelExecutor.ExecutionOutcome outcome = runDag(graphData);
+
+        assertThat(outcome.failed).isTrue();
+        assertThat(outcome.errorMsg).contains("A");
+        assertThat(outcome.finalOutputs).isEmpty();
+        assertThat(bExecutions.get()).as("B 不应被调度").isEqualTo(0);
+        assertThat(cExecutions.get()).as("C 不应被调度").isEqualTo(0);
+        // B、C 应被记录为 SKIPPED
+        assertThat(statusCapture).containsEntry("A", "FAILED")
+                .containsEntry("B", "SKIPPED")
+                .containsEntry("C", "SKIPPED");
+    }
+
+    /**
+     * 场景 3：A→B→C，A 失败 + failStrategy=skip。
+     *
+     * <p>预期：B 和 C 仍被调度执行，workflow 不标记为 failed。</p>
+     */
+    @Test
+    void failedNodeWithSkip_continuesWorkflow() {
+        String graphData = "{"
+                + "\"nodes\":["
+                + "  {\"id\":\"A\",\"type\":\"llm\",\"data\":{\"label\":\"A\",\"failStrategy\":\"skip\"}},"
+                + "  {\"id\":\"B\",\"type\":\"llm\",\"data\":{\"label\":\"B\"}},"
+                + "  {\"id\":\"C\",\"type\":\"llm\",\"data\":{\"label\":\"C\"}}"
+                + "],\"edges\":["
+                + "  {\"source\":\"A\",\"target\":\"B\"},"
+                + "  {\"source\":\"B\",\"target\":\"C\"}"
+                + "]}";
+
+        NodeExecutor llm = mockExecutor("llm", nodeId -> {
+            switch (nodeId) {
+                case "A":
+                    return NodeExecutionResult.failed("A", "故意失败");
+                case "B":
+                    return NodeExecutionResult.success("B", envelope("B-done"));
+                case "C":
+                    return NodeExecutionResult.success("C", envelope("C-done"));
+                default:
+                    return NodeExecutionResult.failed(nodeId, "未知");
+            }
+        });
+        executorsMap.put("llm", llm);
+
+        WorkflowLevelExecutor.ExecutionOutcome outcome = runDag(graphData);
+
+        assertThat(outcome.failed).as("skip 策略不算失败").isFalse();
+        assertThat(statusCapture).containsEntry("A", "FAILED")
+                .containsEntry("B", "SUCCESS")
+                .containsEntry("C", "SUCCESS");
+    }
+
+    /**
+     * 场景 4：A→X→B（branch-1），X→C（branch-2），X 选择 branch-1。
+     *
+     * <p>预期：B 正常执行，C 标记 SKIPPED，C 的 executor 不被调用。</p>
+     */
+    @Test
+    void conditionBranchUnselected_targetMarkedAsSkipped() {
+        String graphData = "{"
+                + "\"nodes\":["
+                + "  {\"id\":\"A\",\"type\":\"llm\",\"data\":{\"label\":\"A\"}},"
+                + "  {\"id\":\"X\",\"type\":\"condition\",\"data\":{\"label\":\"X\"}},"
+                + "  {\"id\":\"B\",\"type\":\"llm\",\"data\":{\"label\":\"B\"}},"
+                + "  {\"id\":\"C\",\"type\":\"llm\",\"data\":{\"label\":\"C\"}}"
+                + "],\"edges\":["
+                + "  {\"source\":\"A\",\"target\":\"X\"},"
+                + "  {\"source\":\"X\",\"target\":\"B\",\"sourceHandle\":\"branch-1\"},"
+                + "  {\"source\":\"X\",\"target\":\"C\",\"sourceHandle\":\"branch-2\"}"
+                + "]}";
+
+        AtomicInteger cExecutions = new AtomicInteger();
+        NodeExecutor llm = mockExecutor("llm", nodeId -> {
+            switch (nodeId) {
+                case "A":
+                    return NodeExecutionResult.success("A", envelope("A-done"));
+                case "B":
+                    return NodeExecutionResult.success("B", envelope("B-done"));
+                case "C":
+                    cExecutions.incrementAndGet();
+                    return NodeExecutionResult.success("C", envelope("C-done"));
+                default:
+                    return NodeExecutionResult.failed(nodeId, "未知");
+            }
+        });
+        NodeExecutor cond = mockExecutor("condition", nodeId ->
+                NodeExecutionResult.success(nodeId, envelope("X-done"), "branch-1"));
+        executorsMap.put("llm", llm);
+        executorsMap.put("condition", cond);
+
+        WorkflowLevelExecutor.ExecutionOutcome outcome = runDag(graphData);
+
+        assertThat(outcome.failed).isFalse();
+        assertThat(statusCapture).containsEntry("A", "SUCCESS")
+                .containsEntry("X", "SUCCESS")
+                .containsEntry("B", "SUCCESS")
+                .containsEntry("C", "SKIPPED");
+        assertThat(cExecutions.get()).as("C 的 executor 不应被调用").isEqualTo(0);
+    }
+
+    /**
+     * 场景 5：A→M, B→M（M 有两个前驱）。A 慢，B 快。
+     *
+     * <p>预期：M 必须在 A 和 B 都完成后才开始执行。
+     * 通过测量 M.startTime >= max(A.endTime, B.endTime) 验证。</p>
+     */
+    @Test
+    void mergeNode_scheduledOnlyAfterAllPredecessorsComplete() throws Exception {
+        String graphData = "{"
+                + "\"nodes\":["
+                + "  {\"id\":\"A\",\"type\":\"llm\",\"data\":{\"label\":\"A\"}},"
+                + "  {\"id\":\"B\",\"type\":\"llm\",\"data\":{\"label\":\"B\"}},"
+                + "  {\"id\":\"M\",\"type\":\"llm\",\"data\":{\"label\":\"M\"}}"
+                + "],\"edges\":["
+                + "  {\"source\":\"A\",\"target\":\"M\"},"
+                + "  {\"source\":\"B\",\"target\":\"M\"}"
+                + "]}";
+
+        CountDownLatch latchA = new CountDownLatch(1);
+        AtomicLong mStartTime = new AtomicLong();
+        AtomicLong aEndTime = new AtomicLong();
+        AtomicLong bEndTime = new AtomicLong();
+        NodeExecutor llm = mockExecutor("llm", nodeId -> {
+            try {
+                switch (nodeId) {
+                    case "A":
+                        Thread.sleep(150);
+                        aEndTime.set(System.currentTimeMillis());
+                        latchA.countDown();
+                        return NodeExecutionResult.success("A", envelope("A-done"));
+                    case "B":
+                        // B 等 A 完成信号后再记录结束时间，确保 A、B 完成时间可比较
+                        latchA.await(2, TimeUnit.SECONDS);
+                        bEndTime.set(System.currentTimeMillis());
+                        return NodeExecutionResult.success("B", envelope("B-done"));
+                    case "M":
+                        mStartTime.set(System.currentTimeMillis());
+                        return NodeExecutionResult.success("M", envelope("M-done"));
+                    default:
+                        return NodeExecutionResult.failed(nodeId, "未知");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                return NodeExecutionResult.failed(nodeId, "中断");
+            }
+        });
+        executorsMap.put("llm", llm);
+
+        WorkflowLevelExecutor.ExecutionOutcome outcome = runDag(graphData);
+
+        assertThat(outcome.failed).isFalse();
+        assertThat(statusCapture).containsEntry("A", "SUCCESS")
+                .containsEntry("B", "SUCCESS")
+                .containsEntry("M", "SUCCESS");
+        // M 必须在 A 和 B 都完成后才开始（>= 较晚完成者的结束时间）
+        assertThat(mStartTime.get())
+                .as("M 必须在 A 和 B 都完成后才调度")
+                .isGreaterThanOrEqualTo(Math.max(aEndTime.get(), bEndTime.get()) - 1);
+    }
+}

docs/workflow-dataflow-execution-plan.md

@@ -0,0 +1,211 @@
+# 工作流引擎 BSP → Dataflow 改造实施计划
+
+> **文档目的**：将工作流执行模型从 BSP（层同步）改造为 Dataflow（数据流），使节点完成时立即触发后继节点，不再被同层慢节点阻塞。
+>
+> **改造范围**：`backend/src/main/java/com/agent/management/engine/WorkflowLevelExecutor.java` 一个文件
+>
+> **状态**：实施中
+
+## 1. 问题确认
+
+### 1.1 用户观察
+
+> 一个节点，后继跟随多个并行节点时，多个并发节点目前是一起完成的。但应该是有快有慢。先完成的节点，又可以紧跟着触发它的后继节点，不必等其他并行节点一起。
+
+### 1.2 根因定位
+
+`WorkflowLevelExecutor.java:162`：
+
+```java
+// 等待当前层所有并行节点完成
+CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
+```
+
+当前执行模型是 **BSP（Bulk Synchronous Parallel，层同步）**：
+- 按 `dag.getLevels()` 逐层遍历
+- 同层节点用 `nodeExecutor` 线程池并行
+- **barrier**：整层必须全部完成才进入下一层
+- 后继节点的调度时机由"所在层"决定，不由"前驱完成时刻"决定
+
+### 1.3 仓库历史
+
+- 从未出现过 dataflow 模型
+- commit `94a9f24` 仅把"层内串行"升级为"层内并行 + barrier"
+- 本次为全新实现
+
+## 2. 改造目标
+
+### 2.1 行为目标
+
+| 模型 | 调度时机 | 总耗时（A→B+C→D，B 5s、C 1s、D 1s） |
+|---|---|---|
+| BSP（当前） | 按层 | A(1s) + max(B,C)(5s) + D(1s) = 7s |
+| Dataflow（目标） | 前驱完成即触发 | A(1s) + max(B, C+D)(5s) = 6s |
+
+更关键的差别：dataflow 下，C 完成的瞬间 D 立刻开始（不等 B）。
+
+### 2.2 必须保留的语义
+
+| 编号 | 现有行为 | 保留方式 |
+|---|---|---|
+| S1 | `executeOneNode()` worker 逻辑 | 完全不动 |
+| S2 | 条件分支 `sourceHandle` 过滤 | 在后继激活中保留 |
+| S3 | `NodeWorkspaceBuilder.build(..., activeEdges)` 可达前驱过滤 | activeEdges 改并发集合后语义不变 |
+| S4 | output 节点的 envelope 写 finalOutputs | 改为持锁写入 |
+| S5 | `sortOrder` 单调递增 | 改为 `AtomicInteger`，按完成顺序 |
+| S6 | failStrategy = abort / skip 语义 | abort 时设置标志 + 取消在飞 future；skip 正常调度下游 |
+| S7 | skipped 节点的 SSE 推送与记录 | 改由"入度永远到不了 0"判定 |
+| S8 | SSE 线程安全（`safeSend` synchronized） | 不动 |
+
+## 3. 设计选型
+
+### 方案 A：CompletableFuture 回调链（**已选**）
+
+每个节点用 `supplyAsync(executeOneNode, nodeExecutor)` 提交；`whenComplete` 中递减后继入度，归零则立即调度后继。
+
+**优点**：
+- 真正满足"前驱完成即触发"
+- 改动聚焦于 `executeByLevel()` 内部
+- `executeOneNode()` / 线程池 / `safeSend` 全部零改动
+- 可通过 `Map<nodeId, Future>` cancel 在飞 future
+
+### 方案 B：Actor + 队列（未选）
+过重，引入新并发原语。
+
+### 方案 C：移除 barrier 但保留 for-level（未选）
+本质仍按层延迟，不满足需求。
+
+## 4. 实施阶段
+
+### 阶段 1：并发数据结构准备
+
+**目标**：BSP 时代的 `HashSet` 升级为并发集合，新增运行时入度计数器；BSP 行为不变。
+
+**改动点**：
+- `activeNodes` / `activeEdges` → `ConcurrentHashMap.newKeySet()`
+- 新增 `Map<String, AtomicInteger> pendingInputs`：运行时入度
+- 新增 `Map<String, CompletableFuture<NodeExecutionResult>> inFlightFutures`：abort cancel 用
+- 新增 `AtomicInteger sortOrderSeq`、`AtomicBoolean abortFlag`、`AtomicReference<String> abortMsgRef`
+
+### 阶段 2：核心调度循环重写
+
+**目标**：删除 `for (level)` + `allOf().join()`，改为事件驱动。
+
+**新增方法**：
+- `scheduleNode(nodeId)`：检查 abortFlag → 提交线程池 → 挂 `whenComplete` 回调
+- `onNodeCompleted(nodeId, result, ex)`：处理结果 → 持锁写 records/output → 触发后继
+- `tryScheduleSuccessors(nodeId, result)`：递减后继入度，归零则调度
+
+**关键顺序**（保证可见性，陷阱 8）：
+```
+activeEdges.add(edge)   ← 1. 先加边
+pendingInputs.decrement ← 2. 再递减入度（happens-before）
+if (入度 == 0) schedule ← 3. 入度归零才调度
+```
+
+**条件分支入度补偿**（陷阱 3/10）：
+- 条件节点 X 有 2 个分支，未选中分支的目标节点 T 入度已在阶段 1 算入
+- X 完成时必须对未选中分支的所有 target 主动递减一次入度，否则 T 永远等不到
+
+**主线程等待**：
+- `AtomicInteger remainingNodes`（初始 = 总节点数）
+- 每个节点完成（含 skipped 路径）递减，归零时 `complete(done)`
+- 主线程 `done.join()`
+
+### 阶段 3：失败策略与 abort 取消
+
+**skip 策略**：与成功一致，下游通过 envelope.status 感知失败
+
+**abort 策略**：
+- `abortFlag.compareAndSet(false, true)` CAS 去重，仅首个失败节点推 `workflow_error`
+- 遍历 `inFlightFutures` 调用 `cancel(true)`
+- `scheduleNode` 与 `tryScheduleSuccessors` 入口检查 abortFlag，未启动节点不再调度
+- 在飞节点完成后**仅清理 inFlightFutures**，不写 records、不推 SSE、不激活下游（避免与已发的 workflow_error 乱序）
+
+### 阶段 4：sortOrder / finalOutputs / SSE 治理
+
+- `sortOrder`：完成顺序，`AtomicInteger.getAndIncrement()` 在持锁区域
+- `finalOutputs`：`clear() + putAll()` 在持锁区域原子执行
+- `nodeRecords`：持锁 + 普通 ArrayList
+- `workflow_error`：阶段 3 CAS 已保证只触发一次
+
+### 阶段 5：单元测试
+
+新增 `WorkflowLevelExecutorTest.java`，覆盖 10 个场景：
+
+| # | 用例 | 验证点 |
+|---|---|---|
+| 1 | 菱形 DAG（A→B+C→D），B 慢 C 快 | D 在 C 完成后立即调度 |
+| 2 | 纯串行 A→B→C | 与 BSP 行为一致 |
+| 3 | 单层 8 节点并行 | sortOrder 唯一单调 |
+| 4 | 中间节点 abort | 下游不调度，workflow_error 一次 |
+| 5 | 中间节点 skip | 下游通过 envelope 感知，仍执行 |
+| 6 | 条件分支未选中 | 未选分支目标节点入度补偿 |
+| 7 | merge 节点多入度 | 等所有 active 前驱完成 |
+| 8 | 多 output 并发 | finalOutputs 原子写入 |
+| 9 | 100 节点 fan-out 压测 | sortOrder 0..99 全覆盖 |
+| 10 | abort 时在飞节点 | 在飞完成但不记录，下游不激活 |
+
+### 阶段 6：回归验证
+
+- 手动跑 3-5 个典型工作流，对比 finalOutputs / nodeRecords
+- 前端兼容性：检查 `WorkflowEditor.vue` / `RunHistory.vue` 是否依赖 sortOrder 全局顺序
+- 故意失败场景验证 abort / skip
+- 16+ 节点并发压测，评估 `max(8, CPU*2)` 线程池容量
+
+## 5. 关键陷阱与对应阶段
+
+| 陷阱 | 描述 | 处理阶段 |
+|---|---|---|
+| 1 | activeNodes / activeEdges 改并发集合 | 阶段 1 |
+| 2 | 新增运行时入度计数器 | 阶段 1 |
+| 3 | 条件分支入度补偿 | 阶段 2 |
+| 4 | inFlightFutures Map 用于 cancel | 阶段 1 + 3 |
+| 5 | sortOrder 全局原子递增 | 阶段 1 + 4 |
+| 6 | finalOutputs 并发写入 | 阶段 4 |
+| 7 | workflow_error CAS 去重 | 阶段 3 |
+| 8 | activeEdges 实时可见性 | 阶段 2 |
+| 9 | 线程池容量评估 | 阶段 6 |
+| 10 | 条件分支 vs 普通节点入度去重对齐 | 阶段 2 |
+| 11 | abort 阻止已就绪下游调度 | 阶段 3 |
+
+## 6. 风险评估
+
+### 高风险（人工 review）
+- **H1 activeEdges 可见性**：worker 启动时必须看到刚 add 的 edge
+- **H2 条件分支入度补偿**：未选中分支目标节点入度必须主动递减，否则死锁
+- **H3 abort 时在飞节点副作用**：完成后不能写 context / 推 SSE / 激活下游
+- **H4 死锁风险**：done 加超时（30 分钟）兜底
+
+### 中风险（测试覆盖）
+- sortOrder 改完成顺序后前端展示
+- finalOutputs 多 output 并发语义
+- 线程池容量
+
+## 7. 不在本次范围内
+
+1. ❌ 不引入 `executionModel = bsp|dataflow` 配置开关
+2. ❌ 不修改 `DagResolver` 分层算法
+3. ❌ 不修改 `NodeExecutor` 接口
+4. ❌ 不引入新依赖（无 Akka / RxJava）
+5. ❌ 不修改 `executeOneNode()` worker 逻辑
+6. ❌ 不扩容线程池
+7. ❌ 不改前端
+
+## 8. 回滚预案
+
+- 改造集中单文件单方法 → `git revert` 原子回滚
+- 无 DB schema 改动
+- 建议阶段 2 与阶段 5 一起合并验证
+
+## 9. 复杂度估算
+
+| 阶段 | 内容 | 估算（小时） |
+|---|---|---|
+| 1 | 并发数据结构准备 | 1.5 - 2 |
+| 2 | 核心调度循环重写 | 3 - 4 |
+| 3 | 失败策略与 abort 取消 | 2.5 - 3 |
+| 4 | sortOrder / finalOutputs / SSE 治理 | 1.5 - 2 |
+| 5 | 单元测试（10 用例） | 3 - 4 |
+| 6 | 回归验证 | 2 - 3 |
+| **合计** | | **13.5 - 18** |

prompt.md

@@ -1306,5 +1306,12 @@ java.lang.IllegalStateException: rag-ai-bridge is disabled
 
 使用技能 git-commit-log-gen 生成提交日志，可以参考 @prompt.md 中的末尾部分需求。
 
+从用户角度对提交日志进行简化，对用户和测试不可见的实现细节不写到日志里。
+
 ---
 
+将上述改造实施计划输出到 @docs 下，然后开始实施。
+
+1. 我扔你是因为你当时笑得不可开交，然后把萌萌叫过去了，我一想就知道你要问她会不会摸老师奶头头，本来萌萌就老跟别人提这事儿，还强化她这个行为
+2. 放屁这个我之前早问过你，跟孕期没关系，你当时在地铁上熏得人直皱眉头的时候怀着孕呢？
+3. 婚礼这个，我也是纠结了很久，一直到周三小萌好差不多才给人家去的答复，主要也是想看看航天那个室友最近咋样