feat: M28 支持工作流多入边汇聚模式

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java

@@ -634,7 +634,8 @@ public class Chain {
                                         NodeStateField
                                                 .EXECUTION_ATTEMPT_KEY);
                             }
-                            if (node.getCondition() == null) {
+                            if (node.getJoinMode() == NodeJoinMode.ANY
+                                    && node.getCondition() == null) {
                                 s.recordTrigger(triggerEdgeId);
                                 fields.add(NodeStateField.TRIGGER_COUNT);
                                 fields.add(NodeStateField.TRIGGER_EDGE_IDS);
@@ -795,7 +796,7 @@ public class Chain {
 
     private boolean shouldSkipNode(Node node, String edgeId) {
         NodeCondition condition = node.getCondition();
-        if (condition == null) {
+        if (node.getJoinMode() == NodeJoinMode.ANY && condition == null) {
             return false;
         }
         return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS, () -> {
@@ -805,7 +806,11 @@ public class Chain {
             });
 
             Map<String, Object> prevResult = Collections.emptyMap();
-            boolean shouldSkipNode = !condition.check(this, newState, prevResult);
+            boolean joinPending = node.getJoinMode() == NodeJoinMode.ALL
+                    && !newState.isUpstreamFullyExecuted();
+            boolean shouldSkipNode = joinPending
+                    || (condition != null
+                    && !condition.check(this, newState, prevResult));
             if (shouldSkipNode) {
                 updateStateSafely(state -> {
                     return state.addUncheckedNodeId(node.id)

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Node.java

@@ -42,6 +42,7 @@ public abstract class Node implements Serializable {
 
     protected NodeCondition condition;
     protected NodeValidator validator;
+    protected NodeJoinMode joinMode = NodeJoinMode.ANY;
 
     // 循环执行相关属性
     protected boolean loopEnable = false;           // 是否启用循环执行
@@ -70,6 +71,10 @@ public abstract class Node implements Serializable {
     }
 
     public void setParentId(String parentId) {
+        if (StringUtil.hasText(parentId) && getJoinMode() == NodeJoinMode.ALL) {
+            throw new IllegalArgumentException(
+                    "joinMode 'all' is not supported for loop child nodes");
+        }
         this.parentId = parentId;
     }
 
@@ -121,6 +126,31 @@ public abstract class Node implements Serializable {
         this.validator = validator;
     }
 
+    /**
+     * 获取节点的直接入边汇聚模式。
+     *
+     * <p>旧序列化对象缺少该字段时返回 {@link NodeJoinMode#ANY}。</p>
+     *
+     * @return 汇聚模式
+     */
+    public NodeJoinMode getJoinMode() {
+        return joinMode == null ? NodeJoinMode.ANY : joinMode;
+    }
+
+    /**
+     * 设置节点的直接入边汇聚模式。
+     *
+     * @param joinMode 汇聚模式
+     */
+    public void setJoinMode(NodeJoinMode joinMode) {
+        NodeJoinMode resolved = joinMode == null ? NodeJoinMode.ANY : joinMode;
+        if (resolved == NodeJoinMode.ALL && StringUtil.hasText(parentId)) {
+            throw new IllegalArgumentException(
+                    "joinMode 'all' is not supported for loop child nodes");
+        }
+        this.joinMode = resolved;
+    }
+
 //    protected void addOutwardEdge(Edge edge) {
 //        if (this.outwardEdges == null) {
 //            this.outwardEdges = new ArrayList<>();

easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/NodeJoinMode.java

@@ -0,0 +1,60 @@
+/**
+ * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
+ * <p>
+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.gnu.org/licenses/lgpl-3.0.txt
+ * <p>
+ * 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 com.easyagents.flow.core.chain;
+
+import java.util.Locale;
+
+/**
+ * 多入边节点的触发汇聚模式。
+ */
+public enum NodeJoinMode {
+
+    /** 任意一条直接入边到达即可执行。 */
+    ANY("any"),
+    /** 全部直接入边到达后才执行。 */
+    ALL("all");
+
+    private final String value;
+
+    NodeJoinMode(String value) {
+        this.value = value;
+    }
+
+    public String getValue() {
+        return value;
+    }
+
+    /**
+     * 按工作流 JSON 值解析汇聚模式。
+     *
+     * @param value 配置值
+     * @return 汇聚模式
+     * @throws IllegalArgumentException 配置为空或不受支持
+     */
+    public static NodeJoinMode ofValue(String value) {
+        if (value == null || value.trim().isEmpty()) {
+            throw new IllegalArgumentException("joinMode must be 'any' or 'all'");
+        }
+        String normalized = value.trim().toLowerCase(Locale.ROOT);
+        for (NodeJoinMode mode : values()) {
+            if (mode.value.equals(normalized)) {
+                return mode;
+            }
+        }
+        throw new IllegalArgumentException(
+                "Unsupported joinMode: " + value + "; expected 'any' or 'all'");
+    }
+}

easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/BaseNodeParser.java

@@ -21,6 +21,7 @@ import com.alibaba.fastjson.JSONObject;
 import com.easyagents.flow.core.chain.DataType;
 import com.easyagents.flow.core.chain.JsCodeCondition;
 import com.easyagents.flow.core.chain.Node;
+import com.easyagents.flow.core.chain.NodeJoinMode;
 import com.easyagents.flow.core.chain.Parameter;
 import com.easyagents.flow.core.chain.RefType;
 import com.easyagents.flow.core.node.BaseNode;
@@ -123,6 +124,10 @@ public abstract class BaseNodeParser<T extends BaseNode> implements NodeParser<T
 
             if (!data.isEmpty()) {
 
+                if (data.containsKey("joinMode")) {
+                    node.setJoinMode(NodeJoinMode.ofValue(data.getString("joinMode")));
+                }
+
                 addParameters(node, data);
                 addOutputDefs(node, data);
 

easy-agents-flow/src/test/java/com/easyagents/flow/core/test/NodeJoinModeTest.java

@@ -0,0 +1,407 @@
+package com.easyagents.flow.core.test;
+
+import com.alibaba.fastjson.JSONObject;
+import com.easyagents.flow.core.chain.Chain;
+import com.easyagents.flow.core.chain.ChainDefinition;
+import com.easyagents.flow.core.chain.ChainState;
+import com.easyagents.flow.core.chain.ChainStatus;
+import com.easyagents.flow.core.chain.Edge;
+import com.easyagents.flow.core.chain.Node;
+import com.easyagents.flow.core.chain.NodeCondition;
+import com.easyagents.flow.core.chain.NodeJoinMode;
+import com.easyagents.flow.core.chain.NodeState;
+import com.easyagents.flow.core.chain.Parameter;
+import com.easyagents.flow.core.chain.RefType;
+import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository;
+import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
+import com.easyagents.flow.core.chain.runtime.ChainExecutor;
+import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
+import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
+import com.easyagents.flow.core.node.BaseNode;
+import com.easyagents.flow.core.node.EndNode;
+import com.easyagents.flow.core.node.StartNode;
+import com.easyagents.flow.core.parser.ChainParser;
+import com.easyagents.flow.core.parser.impl.EndNodeParser;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+
+/**
+ * 验证普通节点的直接入边汇聚模式。
+ */
+public class NodeJoinModeTest {
+
+    @Test
+    public void shouldParseJoinModeAndRejectInvalidValues() {
+        ChainParser parser = ChainParser.builder()
+                .withDefaultParsers(true)
+                .build();
+
+        Assert.assertEquals(
+                NodeJoinMode.ANY,
+                parseEndNode(parser, null, null).getJoinMode());
+        Assert.assertEquals(
+                NodeJoinMode.ALL,
+                parseEndNode(parser, "all", null).getJoinMode());
+        Assert.assertEquals(
+                NodeJoinMode.ANY,
+                parseEndNode(parser, "ANY", null).getJoinMode());
+
+        assertInvalidJoinMode(() -> parseEndNode(parser, "first", null));
+        assertInvalidJoinMode(() -> parseEndNode(parser, "", null));
+        assertInvalidJoinMode(() -> parseEndNode(parser, "all", "loop"));
+
+        ProbeJoinNode node = new ProbeJoinNode(new AtomicInteger());
+        node.setJoinMode(NodeJoinMode.ALL);
+        assertInvalidJoinMode(() -> node.setParentId("loop"));
+    }
+
+    @Test
+    public void shouldWaitForEveryInboundEdgeBeforeExecutingAndCheckingCondition()
+            throws Exception {
+        JoinFixture fixture = createJoinFixture(NodeJoinMode.ALL, true);
+        String instanceId = null;
+        try {
+            instanceId = fixture.executor.executeAsync(
+                    fixture.definition.getId(), Collections.emptyMap());
+            Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS));
+            Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS));
+
+            String currentInstanceId = instanceId;
+            await(() -> hasTriggerEdge(
+                    fixture.nodeStateRepository,
+                    currentInstanceId,
+                    "join",
+                    "a-join"));
+            Assert.assertEquals(0, fixture.joinExecutions.get());
+            Assert.assertEquals(0, fixture.conditionChecks.get());
+
+            fixture.releaseBranchB.countDown();
+            ChainState finalState = awaitTerminal(
+                    fixture.chainStateRepository, instanceId);
+
+            Assert.assertEquals(1, fixture.joinExecutions.get());
+            Assert.assertEquals(1, fixture.conditionChecks.get());
+            Assert.assertEquals("A+B", finalState.getExecuteResult().get("combined"));
+            Assert.assertEquals(Boolean.TRUE, finalState.getExecuteResult().get("sawBoth"));
+        } finally {
+            fixture.releaseBranchB.countDown();
+            fixture.scheduler.shutdown();
+        }
+    }
+
+    @Test
+    public void shouldKeepAnyModeFirstArrivalBehavior() throws Exception {
+        JoinFixture fixture = createJoinFixture(NodeJoinMode.ANY, false);
+        try {
+            String instanceId = fixture.executor.executeAsync(
+                    fixture.definition.getId(), Collections.emptyMap());
+            Assert.assertTrue(fixture.branchACompleted.await(2, TimeUnit.SECONDS));
+            Assert.assertTrue(fixture.branchBStarted.await(2, TimeUnit.SECONDS));
+
+            await(() -> fixture.joinExecutions.get() > 0);
+            ChainState finalState = awaitTerminal(
+                    fixture.chainStateRepository, instanceId);
+
+            Assert.assertEquals(1, fixture.joinExecutions.get());
+            Assert.assertEquals(Boolean.FALSE, finalState.getExecuteResult().get("sawBoth"));
+        } finally {
+            fixture.releaseBranchB.countDown();
+            fixture.scheduler.shutdown();
+        }
+    }
+
+    @Test
+    public void shouldKeepDefaultAnyRetryAndLoopBehavior() throws Exception {
+        ScheduledExecutorService schedulerPool =
+                Executors.newSingleThreadScheduledExecutor();
+        ExecutorService workerPool = Executors.newFixedThreadPool(3);
+        TriggerScheduler scheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L);
+        AtomicInteger executions = new AtomicInteger();
+        ChainDefinition definition = new ChainDefinition();
+        definition.setId("join-mode-retry-loop");
+        StartNode start = new StartNode();
+        start.setId("start");
+        RetryLoopNode worker = new RetryLoopNode(executions);
+        worker.setId("worker");
+        worker.setRetryEnable(true);
+        worker.setMaxRetryCount(1);
+        worker.setRetryIntervalMs(0L);
+        worker.setLoopEnable(true);
+        worker.setMaxLoopCount(2);
+        worker.setLoopIntervalMs(0L);
+        EndNode end = endNode("worker", "count", "count");
+        definition.addNode(start);
+        definition.addNode(worker);
+        definition.addNode(end);
+        definition.addEdge(edge("start-worker", "start", "worker"));
+        definition.addEdge(edge("worker-end", "worker", "end"));
+
+        ChainExecutor executor = new ChainExecutor(
+                ignored -> definition,
+                new InMemoryChainStateRepository(),
+                new InMemoryNodeStateRepository(),
+                scheduler);
+        try {
+            Map<String, Object> result = executor.execute(
+                    definition.getId(), Collections.emptyMap(), 5L, TimeUnit.SECONDS);
+
+            Assert.assertEquals(NodeJoinMode.ANY, worker.getJoinMode());
+            Assert.assertEquals(3, executions.get());
+            Assert.assertEquals(3, result.get("count"));
+        } finally {
+            scheduler.shutdown();
+        }
+    }
+
+    private JoinFixture createJoinFixture(
+            NodeJoinMode joinMode, boolean withCondition) {
+        JoinFixture fixture = new JoinFixture();
+        fixture.schedulerPool = Executors.newScheduledThreadPool(2);
+        fixture.workerPool = Executors.newFixedThreadPool(4);
+        fixture.scheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(),
+                fixture.schedulerPool,
+                fixture.workerPool,
+                1000L);
+        fixture.chainStateRepository = new InMemoryChainStateRepository();
+        fixture.nodeStateRepository = new InMemoryNodeStateRepository();
+        fixture.definition = new ChainDefinition();
+        fixture.definition.setId("join-mode-" + joinMode.getValue());
+
+        StartNode start = new StartNode();
+        start.setId("start");
+        BranchNode branchA = new BranchNode(
+                "A", fixture.branchACompleted, null, null);
+        branchA.setId("a");
+        BranchNode branchB = new BranchNode(
+                "B", null, fixture.branchBStarted, fixture.releaseBranchB);
+        branchB.setId("b");
+        ProbeJoinNode join = new ProbeJoinNode(fixture.joinExecutions);
+        join.setId("join");
+        join.setJoinMode(joinMode);
+        if (withCondition) {
+            join.setCondition(new BothOutputsCondition(fixture.conditionChecks));
+        }
+        EndNode end = endNode("join", "combined", "combined");
+        end.addOutputDef(outputRef("sawBoth", "join.sawBoth"));
+
+        fixture.definition.addNode(start);
+        fixture.definition.addNode(branchA);
+        fixture.definition.addNode(branchB);
+        fixture.definition.addNode(join);
+        fixture.definition.addNode(end);
+        fixture.definition.addEdge(edge("start-a", "start", "a"));
+        fixture.definition.addEdge(edge("start-b", "start", "b"));
+        fixture.definition.addEdge(edge("a-join", "a", "join"));
+        fixture.definition.addEdge(edge("b-join", "b", "join"));
+        fixture.definition.addEdge(edge("join-end", "join", "end"));
+
+        fixture.executor = new ChainExecutor(
+                ignored -> fixture.definition,
+                fixture.chainStateRepository,
+                fixture.nodeStateRepository,
+                fixture.scheduler);
+        return fixture;
+    }
+
+    private Node parseEndNode(
+            ChainParser parser, String joinMode, String parentId) {
+        JSONObject data = new JSONObject();
+        if (joinMode != null) {
+            data.put("joinMode", joinMode);
+        }
+        JSONObject nodeJson = new JSONObject();
+        nodeJson.put("id", "end");
+        nodeJson.put("type", "endNode");
+        nodeJson.put("parentId", parentId);
+        nodeJson.put("data", data);
+        return new EndNodeParser().parse(
+                nodeJson, new JSONObject(), parser);
+    }
+
+    private void assertInvalidJoinMode(Runnable action) {
+        try {
+            action.run();
+            Assert.fail("Expected invalid join mode");
+        } catch (IllegalArgumentException expected) {
+            Assert.assertTrue(expected.getMessage().contains("joinMode"));
+        }
+    }
+
+    private static boolean hasTriggerEdge(
+            InMemoryNodeStateRepository repository,
+            String instanceId,
+            String nodeId,
+            String edgeId) {
+        NodeState state = repository.load(instanceId, nodeId);
+        return state != null && state.getTriggerEdgeIds().contains(edgeId);
+    }
+
+    private static ChainState awaitTerminal(
+            InMemoryChainStateRepository repository,
+            String instanceId) throws Exception {
+        await(() -> {
+            ChainState state = repository.load(instanceId);
+            return state != null
+                    && state.getStatus() != null
+                    && state.getStatus().isTerminal();
+        });
+        ChainState state = repository.load(instanceId);
+        Assert.assertEquals(ChainStatus.SUCCEEDED, state.getStatus());
+        return state;
+    }
+
+    private static void await(BooleanSupplier condition) throws Exception {
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3L);
+        while (!condition.getAsBoolean() && System.nanoTime() < deadline) {
+            Thread.sleep(10L);
+        }
+        Assert.assertTrue("condition was not met before timeout", condition.getAsBoolean());
+    }
+
+    private static EndNode endNode(
+            String sourceNodeId, String sourceName, String outputName) {
+        EndNode end = new EndNode();
+        end.setId("end");
+        end.addOutputDef(outputRef(
+                outputName, sourceNodeId + "." + sourceName));
+        return end;
+    }
+
+    private static Parameter outputRef(String name, String ref) {
+        Parameter parameter = new Parameter();
+        parameter.setName(name);
+        parameter.setRef(ref);
+        parameter.setRefType(RefType.REF);
+        return parameter;
+    }
+
+    private static Edge edge(String id, String source, String target) {
+        Edge edge = new Edge();
+        edge.setId(id);
+        edge.setSource(source);
+        edge.setTarget(target);
+        return edge;
+    }
+
+    private static final class BranchNode extends BaseNode {
+        private final String value;
+        private final CountDownLatch completed;
+        private final CountDownLatch started;
+        private final CountDownLatch release;
+
+        private BranchNode(
+                String value,
+                CountDownLatch completed,
+                CountDownLatch started,
+                CountDownLatch release) {
+            this.value = value;
+            this.completed = completed;
+            this.started = started;
+            this.release = release;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            if (started != null) {
+                started.countDown();
+            }
+            if (release != null) {
+                try {
+                    if (!release.await(3L, TimeUnit.SECONDS)) {
+                        throw new IllegalStateException("branch release timed out");
+                    }
+                } catch (InterruptedException error) {
+                    Thread.currentThread().interrupt();
+                    throw new IllegalStateException("branch interrupted", error);
+                }
+            }
+            if (completed != null) {
+                completed.countDown();
+            }
+            return Collections.singletonMap("value", value);
+        }
+    }
+
+    private static final class ProbeJoinNode extends BaseNode {
+        private final AtomicInteger executions;
+
+        private ProbeJoinNode(AtomicInteger executions) {
+            this.executions = executions;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            executions.incrementAndGet();
+            Object a = chain.getExecutionState().getMemory().get("a.value");
+            Object b = chain.getExecutionState().getMemory().get("b.value");
+            Map<String, Object> result = new HashMap<>();
+            result.put("combined", String.valueOf(a) + "+" + String.valueOf(b));
+            result.put("sawBoth", a != null && b != null);
+            return result;
+        }
+    }
+
+    private static final class BothOutputsCondition implements NodeCondition {
+        private static final long serialVersionUID = 1L;
+        private final AtomicInteger checks;
+
+        private BothOutputsCondition(AtomicInteger checks) {
+            this.checks = checks;
+        }
+
+        @Override
+        public boolean check(
+                Chain chain,
+                NodeState context,
+                Map<String, Object> executeResult) {
+            checks.incrementAndGet();
+            Map<String, Object> memory = chain.getExecutionState().getMemory();
+            return memory.containsKey("a.value") && memory.containsKey("b.value");
+        }
+    }
+
+    private static final class RetryLoopNode extends BaseNode {
+        private final AtomicInteger executions;
+
+        private RetryLoopNode(AtomicInteger executions) {
+            this.executions = executions;
+        }
+
+        @Override
+        public Map<String, Object> execute(Chain chain) {
+            int count = executions.incrementAndGet();
+            if (count == 1) {
+                throw new IllegalStateException("retry once");
+            }
+            return Collections.singletonMap("count", count);
+        }
+    }
+
+    private static final class JoinFixture {
+        private final CountDownLatch branchACompleted = new CountDownLatch(1);
+        private final CountDownLatch branchBStarted = new CountDownLatch(1);
+        private final CountDownLatch releaseBranchB = new CountDownLatch(1);
+        private final AtomicInteger joinExecutions = new AtomicInteger();
+        private final AtomicInteger conditionChecks = new AtomicInteger();
+        private ScheduledExecutorService schedulerPool;
+        private ExecutorService workerPool;
+        private TriggerScheduler scheduler;
+        private InMemoryChainStateRepository chainStateRepository;
+        private InMemoryNodeStateRepository nodeStateRepository;
+        private ChainDefinition definition;
+        private ChainExecutor executor;
+    }
+}