diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java index a539521..b9a5edf 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Chain.java @@ -1751,16 +1751,91 @@ public class Chain { stateInstanceId, 10L, TimeUnit.SECONDS, - () -> { - ChainState current = - chainStateRepository.load(stateInstanceId); - if (current == null - || current.getStatus() != ChainStatus.SUSPEND) { - return false; - } - resumeSuspended(variables); - return true; - }); + () -> resumeSuspended(variables)); + } + + private void validateResumeVariables( + ChainState state, Map variables) { + List parameters = state.getSuspendForParameters(); + if (parameters == null || parameters.isEmpty()) { + return; + } + Map submitted = variables == null + ? Collections.emptyMap() + : variables; + Set expectedKeys = parameters.stream() + .map(Parameter::getName) + .filter(Objects::nonNull) + .collect(java.util.stream.Collectors.toCollection( + LinkedHashSet::new)); + Set extraKeys = new LinkedHashSet<>(submitted.keySet()); + extraKeys.removeAll(expectedKeys); + if (!extraKeys.isEmpty()) { + throw new ChainResumeException( + "确认参数包含未声明字段"); + } + for (Parameter parameter : parameters) { + String name = parameter.getName(); + Object value = submitted.get(name); + String label = StringUtil.getFirstWithText( + parameter.getFormLabel(), name); + if (!submitted.containsKey(name) || isBlankResumeValue(value)) { + throw new ChainResumeException( + "确认参数[" + label + "]不能为空"); + } + validateResumeOption(parameter, value, label); + } + } + + private void validateResumeOption( + Parameter parameter, Object value, String label) { + List options = parameter.getOptions(); + if (options == null || options.isEmpty()) { + return; + } + Set allowedValues = options.stream() + .map(ParameterOption::getValue) + .filter(Objects::nonNull) + .collect(java.util.stream.Collectors.toCollection( + LinkedHashSet::new)); + if ("checkbox".equals(parameter.getFormType())) { + if (!(value instanceof Collection selected)) { + throw new ChainResumeException( + "确认参数[" + label + "]必须提交字符串数组"); + } + Set unique = new LinkedHashSet<>(); + for (Object item : selected) { + if (!(item instanceof String selectedValue) + || !allowedValues.contains(selectedValue)) { + throw new ChainResumeException( + "确认参数[" + label + "]包含未配置选项"); + } + if (!unique.add(selectedValue)) { + throw new ChainResumeException( + "确认参数[" + label + "]不能重复选择同一选项"); + } + } + return; + } + if (!(value instanceof String selectedValue)) { + throw new ChainResumeException( + "确认参数[" + label + "]必须提交单个字符串值"); + } + if (!allowedValues.contains(selectedValue)) { + throw new ChainResumeException( + "确认参数[" + label + "]包含未配置选项"); + } + } + + private boolean isBlankResumeValue(Object value) { + if (value == null) { + return true; + } + if (value instanceof String text) { + return text.trim().isEmpty(); + } + return value instanceof Collection collection + && collection.isEmpty(); } /** @@ -1777,36 +1852,51 @@ public class Chain { * * @param variables 恢复时注入的变量 */ - private void resumeSuspended(Map variables) { - ChainState newState = updateStateSafely(state -> { - if (variables != null) { - state.getMemory().putAll(variables); - return EnumSet.of(ChainStateField.MEMORY); - } else { + private boolean resumeSuspended(Map variables) { + AtomicBoolean resumed = new AtomicBoolean(false); + AtomicReference> suspendedNodeIds = + new AtomicReference<>(Collections.emptySet()); + updateStateSafely(state -> { + resumed.set(false); + suspendedNodeIds.set(Collections.emptySet()); + if (state.getStatus() != ChainStatus.SUSPEND) { return null; } + validateResumeVariables(state, variables); + if (state.getSuspendNodeIds() != null) { + suspendedNodeIds.set( + new LinkedHashSet<>(state.getSuspendNodeIds())); + } + EnumSet updatedFields = EnumSet.of( + ChainStateField.STATUS, + ChainStateField.SUSPEND_NODE_IDS, + ChainStateField.SUSPEND_FOR_PARAMETERS); + if (variables != null && !variables.isEmpty()) { + state.getMemory().putAll(variables); + updatedFields.add(ChainStateField.MEMORY); + } + state.setStatus(ChainStatus.RUNNING); + state.setSuspendNodeIds(null); + state.setSuspendForParameters(null); + resumed.set(true); + return updatedFields; }); + if (!resumed.get()) { + return false; + } notifyEvent(new ChainResumeEvent(this, variables)); - setStatusAndNotifyEvent(ChainStatus.RUNNING); + notifyEvent(new ChainStatusChangeEvent( + this, ChainStatus.RUNNING, ChainStatus.SUSPEND)); - Set suspendNodeIds = newState.getSuspendNodeIds(); - if (suspendNodeIds != null && !suspendNodeIds.isEmpty()) { - // 移除 suspend 状态，方便二次 suspend 时，不带有旧数据 - updateStateSafely(state -> { - state.setSuspendNodeIds(null); - state.setSuspendForParameters(null); - return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS, ChainStateField.SUSPEND_FOR_PARAMETERS); - }); - - for (String id : suspendNodeIds) { - Node node = definition.getNodeById(id); - if (node == null) { - throw new ChainException("Node not found: " + id); - } - scheduleNode(node, null, TriggerType.RESUME, 0L); + for (String id : suspendedNodeIds.get()) { + Node node = definition.getNodeById(id); + if (node == null) { + throw new ChainException("Node not found: " + id); } + scheduleNode(node, null, TriggerType.RESUME, 0L); } + return true; } public void resume() { diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainResumeException.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainResumeException.java new file mode 100644 index 0000000..ff34bbe --- /dev/null +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ChainResumeException.java @@ -0,0 +1,16 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0. + */ +package com.easyagents.flow.core.chain; + +/** + * 工作流挂起参数不满足当前恢复请求时抛出的异常。 + */ +public class ChainResumeException extends ChainException { + + public ChainResumeException(String message) { + super(message); + } +} diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java index 379629f..0d297e9 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/Parameter.java @@ -49,6 +49,11 @@ public class Parameter implements Serializable, Cloneable { */ protected List enums; + /** + * 显示文案与实际值分离的结构化选项。 + */ + protected List options; + /** * 用户输入的表单类型，例如："input" "textarea" "select" "radio" "checkbox" 等等 */ @@ -242,6 +247,14 @@ public class Parameter implements Serializable, Cloneable { } } + public List getOptions() { + return options; + } + + public void setOptions(List options) { + this.options = options; + } + public String getFormType() { return formType; } @@ -298,6 +311,7 @@ public class Parameter implements Serializable, Cloneable { ", flattenAggregation=" + flattenAggregation + ", children=" + children + ", enums=" + enums + + ", options=" + options + ", formType='" + formType + '\'' + ", formLabel='" + formLabel + '\'' + ", formPlaceholder='" + formPlaceholder + '\'' + @@ -320,6 +334,13 @@ public class Parameter implements Serializable, Cloneable { clone.enums = new ArrayList<>(this.enums.size()); clone.enums.addAll(this.enums); } + if (this.options != null) { + clone.options = new ArrayList<>(this.options.size()); + for (ParameterOption option : this.options) { + clone.options.add(new ParameterOption( + option.getLabel(), option.getValue())); + } + } return clone; } catch (CloneNotSupportedException e) { throw new AssertionError(); diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ParameterOption.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ParameterOption.java new file mode 100644 index 0000000..e84554c --- /dev/null +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/ParameterOption.java @@ -0,0 +1,55 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0. + */ +package com.easyagents.flow.core.chain; + +import java.io.Serializable; + +/** + * 用户输入参数的结构化选项。 + */ +public class ParameterOption implements Serializable { + + private static final long serialVersionUID = 1L; + + private String label; + private String value; + + public ParameterOption() { + } + + public ParameterOption(String label, String value) { + setLabel(label); + setValue(value); + } + + public String getLabel() { + return label; + } + + public void setLabel(String label) { + this.label = trim(label); + } + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = trim(value); + } + + private static String trim(String value) { + return value == null ? null : value.trim(); + } + + @Override + public String toString() { + return "ParameterOption{" + + "label='" + label + '\'' + + ", value='" + value + '\'' + + '}'; + } +} diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/ConfirmNode.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/ConfirmNode.java index c1f17ba..1c02d6d 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/ConfirmNode.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/node/ConfirmNode.java @@ -15,21 +15,55 @@ */ package com.easyagents.flow.core.node; - import com.easyagents.flow.core.chain.Chain; import com.easyagents.flow.core.chain.ChainSuspendException; +import com.easyagents.flow.core.chain.DataType; import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.ParameterOption; import com.easyagents.flow.core.chain.RefType; import com.easyagents.flow.core.chain.repository.ChainStateField; +import com.easyagents.flow.core.chain.runtime.Trigger; +import com.easyagents.flow.core.chain.runtime.TriggerContext; +import com.easyagents.flow.core.chain.runtime.TriggerType; -import java.util.*; +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; public class ConfirmNode extends BaseNode { private static final long serialVersionUID = 1L; + public static final String DEFAULT_OUTPUT_NAME = "selection"; + public static final int MAX_OPTIONS = 100; + public static final int MAX_MESSAGE_LENGTH = 2000; + public static final int MAX_OPTION_LENGTH = 200; + public static final Set SUPPORTED_CONFIGURATION_KEYS = Set.of( + "condition", + "description", + "expand", + "joinMode", + "loopBreakCondition", + "loopEnable", + "loopIntervalMs", + "maxLoopCount", + "maxRetryCount", + "message", + "multiple", + "options", + "outputDefs", + "resetRetryCountAfterNormal", + "retryEnable", + "retryIntervalMs", + "title"); private String message; - private List confirms; + private boolean multiple; + private List options; public String getMessage() { return message; @@ -39,115 +73,152 @@ public class ConfirmNode extends BaseNode { this.message = message; } - public List getConfirms() { - return confirms; + public boolean isMultiple() { + return multiple; } - public void setConfirms(List confirms) { - if (confirms != null) { - for (Parameter confirm : confirms) { - confirm.setRefType(RefType.INPUT); - confirm.setRequired(true); // 必填，才能正确通过 getParameterValuesOnly 获取参数值 - confirm.setName(confirm.getName()); - } - } - this.confirms = confirms; + public void setMultiple(boolean multiple) { + this.multiple = multiple; } + public List getOptions() { + return options; + } + + public void setOptions(List options) { + this.options = options; + } @Override public Map execute(Chain chain) { + validateConfiguration(); + String outputName = resolveOutputName(); + Parameter parameter = buildParameter(); - List confirmParameters = new ArrayList<>(); - addConfirmParameter(confirmParameters); - - if (confirms != null) { - for (Parameter confirm : confirms) { - Parameter clone = confirm.clone(); - clone.setName(confirm.getName() + "__" + getId()); - clone.setRefType(RefType.INPUT); - confirmParameters.add(clone); - } + // 确认值只能来自经过 Chain.resumeIfSuspended 校验后创建的恢复触发器。 + // 启动参数与普通节点内存中的同名值均不能绕过人工确认。 + if (!isValidatedResumeTrigger(chain)) { + chain.updateStateSafely(state -> { + if (state.getMemory().remove(parameter.getName()) == null) { + return null; + } + return EnumSet.of(ChainStateField.MEMORY); + }); } Map values; try { values = chain.getExecutionState() - .resolveParameters(this, confirmParameters); - // 移除 confirm 参数，方便在其他节点二次确认，或者在 for 循环中第二次获取 + .resolveParameters(this, Collections.singletonList(parameter)); chain.updateStateSafely(state -> { - for (Parameter confirmParameter : confirmParameters) { - state.getMemory().remove(confirmParameter.getName()); + if (!state.getMemory().containsKey(parameter.getName())) { + return null; } + state.getMemory().remove(parameter.getName()); return EnumSet.of(ChainStateField.MEMORY); }); - } catch (ChainSuspendException e) { + } catch (ChainSuspendException exception) { chain.updateStateSafely(state -> { state.setMessage(message); return EnumSet.of(ChainStateField.MESSAGE); }); - - if (confirms != null) { - List newParameters = new ArrayList<>(); - for (Parameter confirm : confirms) { - Parameter clone = confirm.clone(); - clone.setName(confirm.getName() + "__" + getId()); - clone.setRefType(RefType.REF); // 固定为 REF - newParameters.add(clone); - } - - // 获取参数值，不会触发 ChainSuspendException 错误 - Map parameterValues = - chain.getExecutionState().resolveParameters( - this, - newParameters, - null, - true); - - // 设置 enums，方便前端给用户进行选择 - for (Parameter confirmParameter : confirmParameters) { - if (confirmParameter.getEnums() == null) { - Object enumsObject = parameterValues.get(confirmParameter.getName()); - confirmParameter.setEnumsObject(enumsObject); - } - } - } - - throw e; + throw exception; } + return Collections.singletonMap( + outputName, + values.get(parameter.getName())); + } - Map results = new HashMap<>(values.size()); - values.forEach((key, value) -> { - int index = key.lastIndexOf("__"); - if (index >= 0) { - results.put(key.substring(0, index), value); - } else { - results.put(key, value); + /** + * 获取并校验当前节点配置的唯一输出名称。 + * + * @return 用户配置的输出名称 + */ + public String resolveOutputName() { + if (outputDefs == null || outputDefs.size() != 1 + || outputDefs.get(0) == null) { + throw new IllegalArgumentException( + "用户确认节点必须配置一个输出参数"); + } + Parameter output = outputDefs.get(0); + String outputName = output.getName(); + if (outputName == null || outputName.trim().isEmpty()) { + throw new IllegalArgumentException( + "用户确认节点输出参数名称不能为空"); + } + DataType expectedType = multiple + ? DataType.Array_String + : DataType.String; + if (output.getDataType() != expectedType) { + throw new IllegalArgumentException( + "用户确认节点输出参数类型必须与选择方式一致"); + } + return outputName; + } + + private boolean isValidatedResumeTrigger(Chain chain) { + Trigger trigger = TriggerContext.getCurrentTrigger(); + return trigger != null + && trigger.getType() == TriggerType.RESUME + && Objects.equals( + chain.getStateInstanceId(), + trigger.getStateInstanceId()) + && Objects.equals(getId(), trigger.getNodeId()); + } + + public void validateConfiguration() { + requireText(message, MAX_MESSAGE_LENGTH, "确认提示内容"); + if (options == null || options.isEmpty()) { + throw new IllegalArgumentException("用户确认节点至少需要一个选项"); + } + if (options.size() > MAX_OPTIONS) { + throw new IllegalArgumentException( + "用户确认节点最多支持 " + MAX_OPTIONS + " 个选项"); + } + + Set normalizedOptions = new HashSet<>(); + for (String option : options) { + String normalized = requireText(option, MAX_OPTION_LENGTH, "选项内容"); + if (!normalizedOptions.add(normalized)) { + throw new IllegalArgumentException("用户确认节点选项内容重复: " + normalized); } - }); - - return results; + } } - - private void addConfirmParameter(List parameters) { - // “确认 和 取消” 的参数 + private Parameter buildParameter() { Parameter parameter = new Parameter(); + parameter.setId(DEFAULT_OUTPUT_NAME); + parameter.setName(DEFAULT_OUTPUT_NAME + "__" + getId()); + parameter.setDataType(multiple ? DataType.Array_String : DataType.String); parameter.setRefType(RefType.INPUT); - parameter.setId("confirm"); - parameter.setName("confirm__" + getId()); parameter.setRequired(true); - - List selectionData = new ArrayList<>(); - selectionData.add("yes"); - selectionData.add("no"); - - parameter.setEnums(selectionData); parameter.setContentType("text"); - parameter.setFormType("confirm"); - parameters.add(parameter); + parameter.setFormType(multiple ? "checkbox" : "radio"); + parameter.setFormLabel("选择内容"); + parameter.setOptions(buildRuntimeOptions()); + return parameter; } + private List buildRuntimeOptions() { + List runtimeOptions = new ArrayList<>(options.size()); + for (String option : options) { + String normalized = option.trim(); + runtimeOptions.add(new ParameterOption(normalized, normalized)); + } + return runtimeOptions; + } + private static String requireText( + String value, int maxLength, String fieldName) { + String normalized = value == null ? "" : value.trim(); + if (normalized.isEmpty()) { + throw new IllegalArgumentException(fieldName + "不能为空"); + } + if (normalized.length() > maxLength) { + throw new IllegalArgumentException( + fieldName + "不能超过 " + maxLength + " 个字符"); + } + return normalized; + } } diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/ConfirmNodeParser.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/ConfirmNodeParser.java index 05bb22b..14fc3ba 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/ConfirmNodeParser.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/parser/impl/ConfirmNodeParser.java @@ -15,11 +15,12 @@ */ package com.easyagents.flow.core.parser.impl; +import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; -import com.easyagents.flow.core.chain.Parameter; import com.easyagents.flow.core.node.ConfirmNode; import com.easyagents.flow.core.parser.BaseNodeParser; +import java.util.ArrayList; import java.util.List; public class ConfirmNodeParser extends BaseNodeParser { @@ -28,12 +29,36 @@ public class ConfirmNodeParser extends BaseNodeParser { public ConfirmNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) { ConfirmNode confirmNode = new ConfirmNode(); - confirmNode.setMessage(data.getString("message")); - - List confirms = getParameters(data, "confirms"); - if (confirms != null && !confirms.isEmpty()) { - confirmNode.setConfirms(confirms); + for (String key : data.keySet()) { + if (!ConfirmNode.SUPPORTED_CONFIGURATION_KEYS.contains(key)) { + throw new IllegalArgumentException( + "用户确认节点包含无效配置字段: " + key); + } } + Object message = data.get("message"); + if (!(message instanceof String)) { + throw new IllegalArgumentException("用户确认节点提示内容必须为字符串"); + } + confirmNode.setMessage((String) message); + + Object multiple = data.get("multiple"); + if (!(multiple instanceof Boolean)) { + throw new IllegalArgumentException("用户确认节点选择方式必须为布尔值"); + } + confirmNode.setMultiple((Boolean) multiple); + + Object optionsValue = data.get("options"); + if (!(optionsValue instanceof JSONArray options)) { + throw new IllegalArgumentException("用户确认节点选项必须为数组"); + } + List confirmOptions = new ArrayList<>(options.size()); + for (Object option : options) { + if (!(option instanceof String)) { + throw new IllegalArgumentException("用户确认节点选项内容必须为字符串"); + } + confirmOptions.add((String) option); + } + confirmNode.setOptions(confirmOptions); return confirmNode; } diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java index 266147d..3a44c53 100644 --- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java @@ -15,10 +15,15 @@ */ package com.easyagents.flow.core.test; +import com.alibaba.fastjson.JSONArray; +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.ChainStatus; +import com.easyagents.flow.core.chain.DataType; import com.easyagents.flow.core.chain.Edge; +import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.RefType; import com.easyagents.flow.core.chain.ChainState; import com.easyagents.flow.core.chain.event.ChainEndEvent; import com.easyagents.flow.core.chain.repository.ChainDefinitionSnapshotRepository; @@ -35,6 +40,7 @@ import com.easyagents.flow.core.node.EndNode; import com.easyagents.flow.core.node.BaseNode; import com.easyagents.flow.core.node.ConfirmNode; import com.easyagents.flow.core.node.StartNode; +import com.easyagents.flow.core.parser.ChainParser; import org.junit.Assert; import org.junit.Test; @@ -61,6 +67,127 @@ import java.util.concurrent.atomic.AtomicReference; */ public class ChainExecutorConcurrencyTest { + /** + * 验证启动变量不能伪造确认节点的恢复参数并绕过人工确认。 + */ + @Test + public void shouldSuspendConfirmNodeDespitePrefilledStartVariable() + throws Exception { + ScheduledExecutorService schedulerPool = + Executors.newSingleThreadScheduledExecutor(); + ExecutorService workerPool = Executors.newFixedThreadPool(2); + TriggerScheduler triggerScheduler = new TriggerScheduler( + new InMemoryTriggerStore(), schedulerPool, workerPool, 10L); + ChainDefinition definition = createConfirmDefinition(); + InMemoryChainStateRepository stateRepository = + new InMemoryChainStateRepository(); + ChainExecutor executor = new ChainExecutor( + ignored -> definition, + stateRepository, + new InMemoryNodeStateRepository(), + triggerScheduler); + + try { + String instanceId = executor.executeAsync( + definition.getId(), + Map.of("selection__confirm", "未配置值")); + + ChainState state = awaitStatus( + stateRepository, instanceId, ChainStatus.SUSPEND); + Assert.assertFalse( + state.getMemory().containsKey("selection__confirm")); + Assert.assertEquals( + "selection__confirm", + state.getSuspendForParameters().get(0).getName()); + } finally { + triggerScheduler.shutdown(); + } + } + + /** + * 验证设计器最终契约可解析、挂起、恢复，并把用户选择按配置名称交给结束节点。 + */ + @Test + public void shouldFlowConfiguredConfirmOutputToEndNode() + throws Exception { + ScheduledExecutorService schedulerPool = + Executors.newSingleThreadScheduledExecutor(); + ExecutorService workerPool = Executors.newFixedThreadPool(2); + TriggerScheduler triggerScheduler = new TriggerScheduler( + new InMemoryTriggerStore(), schedulerPool, workerPool, 10L); + ChainDefinition definition = createParsedConfirmDefinition(); + InMemoryChainStateRepository stateRepository = + new InMemoryChainStateRepository(); + ChainExecutor executor = new ChainExecutor( + ignored -> definition, + stateRepository, + new InMemoryNodeStateRepository(), + triggerScheduler); + + try { + String instanceId = executor.executeAsync( + definition.getId(), Collections.emptyMap()); + ChainState suspended = awaitStatus( + stateRepository, instanceId, ChainStatus.SUSPEND); + + Assert.assertEquals( + "selection__confirm", + suspended.getSuspendForParameters().get(0).getName()); + Assert.assertTrue(executor.resumeAsyncIfSuspended( + instanceId, + Map.of("selection__confirm", "继续"))); + + ChainState completed = awaitStatus( + stateRepository, instanceId, ChainStatus.SUCCEEDED); + Assert.assertEquals("继续", completed.getExecuteResult().get("result")); + } finally { + triggerScheduler.shutdown(); + } + } + + /** + * 验证多选确认结果以字符串数组形式流转到结束节点。 + */ + @Test + public void shouldFlowMultipleConfirmOutputToEndNode() + throws Exception { + ScheduledExecutorService schedulerPool = + Executors.newSingleThreadScheduledExecutor(); + ExecutorService workerPool = Executors.newFixedThreadPool(2); + TriggerScheduler triggerScheduler = new TriggerScheduler( + new InMemoryTriggerStore(), schedulerPool, workerPool, 10L); + ChainDefinition definition = createParsedConfirmDefinition(true); + InMemoryChainStateRepository stateRepository = + new InMemoryChainStateRepository(); + ChainExecutor executor = new ChainExecutor( + ignored -> definition, + stateRepository, + new InMemoryNodeStateRepository(), + triggerScheduler); + + try { + String instanceId = executor.executeAsync( + definition.getId(), Collections.emptyMap()); + ChainState suspended = awaitStatus( + stateRepository, instanceId, ChainStatus.SUSPEND); + + Assert.assertEquals( + DataType.Array_String, + suspended.getSuspendForParameters().get(0).getDataType()); + List selection = List.of("继续", "停止"); + Assert.assertTrue(executor.resumeAsyncIfSuspended( + instanceId, + Map.of("selection__confirm", selection))); + + ChainState completed = awaitStatus( + stateRepository, instanceId, ChainStatus.SUCCEEDED); + Assert.assertEquals( + selection, completed.getExecuteResult().get("result")); + } finally { + triggerScheduler.shutdown(); + } + } + /** * 验证同步 Tool 入口遇到人工挂起会快速失败，不会无限占用调用线程。 * @@ -563,6 +690,12 @@ public class ChainExecutorConcurrencyTest { start.setId("start"); ConfirmNode confirm = new ConfirmNode(); confirm.setId("confirm"); + confirm.setMessage("请选择是否继续"); + confirm.setOptions(List.of("继续", "停止")); + confirm.setOutputDefs(Collections.singletonList( + new Parameter( + ConfirmNode.DEFAULT_OUTPUT_NAME, + DataType.String))); EndNode end = new EndNode(); end.setId("end"); Edge first = new Edge(); @@ -581,6 +714,91 @@ public class ChainExecutorConcurrencyTest { return definition; } + private ChainDefinition createParsedConfirmDefinition() { + return createParsedConfirmDefinition(false); + } + + private ChainDefinition createParsedConfirmDefinition(boolean multiple) { + JSONArray nodes = new JSONArray(); + nodes.add(nodeJson("start", "startNode", new JSONObject())); + + JSONObject confirmData = new JSONObject(); + confirmData.put("message", "请选择是否继续"); + confirmData.put("multiple", multiple); + confirmData.put("options", new JSONArray(List.of("继续", "停止"))); + String outputType = multiple ? "Array" : "String"; + confirmData.put("outputDefs", new JSONArray(List.of( + parameterJson("templateType", outputType, null)))); + nodes.add(nodeJson("confirm", "confirmNode", confirmData)); + + JSONObject endData = new JSONObject(); + endData.put("outputDefs", new JSONArray(List.of( + parameterJson( + "result", outputType, "confirm.templateType")))); + nodes.add(nodeJson("end", "endNode", endData)); + + JSONArray edges = new JSONArray(); + edges.add(edgeJson("start-to-confirm", "start", "confirm")); + edges.add(edgeJson("confirm-to-end", "confirm", "end")); + JSONObject flow = new JSONObject(); + flow.put("nodes", nodes); + flow.put("edges", edges); + + ChainDefinition definition = ChainParser.builder() + .withDefaultParsers(true) + .build() + .parse(flow.toJSONString()); + definition.setId("confirm-output-flow-test"); + return definition; + } + + private JSONObject nodeJson( + String id, String type, JSONObject data) { + JSONObject node = new JSONObject(); + node.put("id", id); + node.put("type", type); + node.put("data", data); + return node; + } + + private JSONObject edgeJson( + String id, String source, String target) { + JSONObject edge = new JSONObject(); + edge.put("id", id); + edge.put("source", source); + edge.put("target", target); + return edge; + } + + private JSONObject parameterJson( + String name, String dataType, String ref) { + JSONObject parameter = new JSONObject(); + parameter.put("name", name); + parameter.put("dataType", dataType); + if (ref != null) { + parameter.put("ref", ref); + parameter.put("refType", RefType.REF.toString()); + } + return parameter; + } + + private ChainState awaitStatus( + InMemoryChainStateRepository repository, + String instanceId, + ChainStatus expected) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + ChainState state; + do { + state = repository.load(instanceId); + if (state != null && state.getStatus() == expected) { + return state; + } + Thread.sleep(10L); + } while (System.nanoTime() < deadline); + Assert.fail("workflow did not reach status " + expected); + return state; + } + /** * 创建用于取消传播验证的工作流。 * diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainResumeGuardTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainResumeGuardTest.java index 6013c34..a57fa5b 100644 --- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainResumeGuardTest.java +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainResumeGuardTest.java @@ -2,15 +2,29 @@ package com.easyagents.flow.core.test; import com.easyagents.flow.core.chain.Chain; import com.easyagents.flow.core.chain.ChainDefinition; +import com.easyagents.flow.core.chain.ChainResumeException; +import com.easyagents.flow.core.chain.ChainState; import com.easyagents.flow.core.chain.ChainStatus; import com.easyagents.flow.core.chain.EventManager; +import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.ParameterOption; +import com.easyagents.flow.core.chain.repository.ChainStateField; import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository; import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository; import org.junit.Assert; import org.junit.Test; +import java.util.Arrays; import java.util.Collections; +import java.util.EnumSet; +import java.util.List; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; /** * {@link Chain} 暂停恢复状态守卫测试。 @@ -72,6 +86,167 @@ public class ChainResumeGuardTest { .containsKey("unexpected")); } + /** + * 验证确认选项只能按挂起时声明的字段和值恢复。 + */ + @Test + public void shouldValidateDeclaredResumeOptionsBeforeStateTransition() { + InMemoryChainStateRepository stateRepository = + new InMemoryChainStateRepository(); + Chain chain = createChain(stateRepository, "resume-options"); + Parameter single = optionParameter( + "templateType__confirm", "会议类型", "radio"); + Parameter multiple = optionParameter( + "participants__confirm", "参会人员", "checkbox"); + chain.getExecutionState().setSuspendForParameters( + Arrays.asList(single, multiple)); + chain.suspend(); + + ChainResumeException invalidOption = assertRejected( + chain, + Map.of( + "templateType__confirm", "UNKNOWN", + "participants__confirm", List.of("REVIEW"))); + ChainResumeException extraField = assertRejected( + chain, + Map.of( + "templateType__confirm", "AGENDA", + "participants__confirm", List.of("REVIEW", "REVIEW"))); + assertRejected( + chain, + Map.of( + "templateType__confirm", "AGENDA", + "participants__confirm", List.of("REVIEW"), + "extra", "value")); + Assert.assertFalse(invalidOption.getMessage().contains("UNKNOWN")); + Assert.assertFalse(extraField.getMessage().contains("extra")); + + Assert.assertEquals( + ChainStatus.SUSPEND, + stateRepository.load("resume-options").getStatus()); + Assert.assertTrue( + stateRepository.load("resume-options").getMemory().isEmpty()); + + boolean resumed = chain.resumeIfSuspended(Map.of( + "templateType__confirm", "AGENDA", + "participants__confirm", List.of("REVIEW", "BRIEFING"))); + + Assert.assertTrue(resumed); + Assert.assertEquals( + "AGENDA", + stateRepository.load("resume-options") + .getMemory() + .get("templateType__confirm")); + Assert.assertEquals( + List.of("REVIEW", "BRIEFING"), + stateRepository.load("resume-options") + .getMemory() + .get("participants__confirm")); + } + + /** + * 验证并发恢复同一暂停实例时，只有一个请求可以完成状态转换。 + */ + @Test + public void shouldAllowOnlyOneConcurrentResume() throws Exception { + InMemoryChainStateRepository stateRepository = + new InMemoryChainStateRepository(); + Chain first = createChain(stateRepository, "resume-concurrent"); + Chain second = createChain(stateRepository, "resume-concurrent"); + Parameter parameter = optionParameter( + "templateType__confirm", "会议类型", "radio"); + first.getExecutionState().setSuspendForParameters( + List.of(parameter)); + first.suspend(); + + CountDownLatch ready = new CountDownLatch(2); + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future firstResult = executor.submit(() -> { + ready.countDown(); + start.await(); + return first.resumeIfSuspended( + Map.of("templateType__confirm", "AGENDA")); + }); + Future secondResult = executor.submit(() -> { + ready.countDown(); + start.await(); + return second.resumeIfSuspended( + Map.of("templateType__confirm", "REVIEW")); + }); + + Assert.assertTrue(ready.await(5, TimeUnit.SECONDS)); + start.countDown(); + int resumedCount = (firstResult.get(5, TimeUnit.SECONDS) ? 1 : 0) + + (secondResult.get(5, TimeUnit.SECONDS) ? 1 : 0); + + Assert.assertEquals(1, resumedCount); + Assert.assertEquals( + ChainStatus.RUNNING, + stateRepository.load("resume-concurrent").getStatus()); + Object selected = stateRepository.load("resume-concurrent") + .getMemory() + .get("templateType__confirm"); + Assert.assertTrue( + "AGENDA".equals(selected) || "REVIEW".equals(selected)); + } finally { + executor.shutdownNow(); + } + } + + /** + * 验证恢复变量、状态和暂停上下文通过一次原子更新完成。 + */ + @Test + public void shouldCommitResumeTransitionInSingleStateUpdate() { + CountingChainStateRepository stateRepository = + new CountingChainStateRepository(); + Chain chain = createChain(stateRepository, "resume-single-update"); + chain.getExecutionState().setSuspendForParameters(List.of( + optionParameter("templateType__confirm", "会议类型", "radio"))); + chain.suspend(); + stateRepository.resetUpdateCount(); + + boolean resumed = chain.resumeIfSuspended( + Map.of("templateType__confirm", "AGENDA")); + + ChainState state = stateRepository.load("resume-single-update"); + Assert.assertTrue(resumed); + Assert.assertEquals(1, stateRepository.getUpdateCount()); + Assert.assertEquals(ChainStatus.RUNNING, state.getStatus()); + Assert.assertNull(state.getSuspendNodeIds()); + Assert.assertNull(state.getSuspendForParameters()); + Assert.assertEquals("AGENDA", state.getMemory().get( + "templateType__confirm")); + } + + private ChainResumeException assertRejected( + Chain chain, Map variables) { + try { + chain.resumeIfSuspended(variables); + Assert.fail("invalid resume variables must be rejected"); + return null; + } catch (ChainResumeException expected) { + Assert.assertNotNull(expected.getMessage()); + return expected; + } + } + + private Parameter optionParameter( + String name, String label, String formType) { + Parameter parameter = new Parameter(); + parameter.setName(name); + parameter.setFormLabel(label); + parameter.setFormType(formType); + parameter.setRequired(true); + parameter.setOptions(List.of( + new ParameterOption("第一议题", "AGENDA"), + new ParameterOption("审议类", "REVIEW"), + new ParameterOption("听取类", "BRIEFING"))); + return parameter; + } + /** * 创建使用进程内状态仓储的最小工作流。 * @@ -93,4 +268,25 @@ public class ChainResumeGuardTest { chain.setEventManager(new EventManager()); return chain; } + + private static class CountingChainStateRepository + extends InMemoryChainStateRepository { + private final AtomicInteger updateCount = new AtomicInteger(); + + @Override + public boolean tryUpdate( + ChainState chainState, + EnumSet fields) { + updateCount.incrementAndGet(); + return super.tryUpdate(chainState, fields); + } + + private int getUpdateCount() { + return updateCount.get(); + } + + private void resetUpdateCount() { + updateCount.set(0); + } + } } diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ConfirmNodeTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ConfirmNodeTest.java new file mode 100644 index 0000000..ba979eb --- /dev/null +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ConfirmNodeTest.java @@ -0,0 +1,235 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0. + */ +package com.easyagents.flow.core.test; + +import com.alibaba.fastjson.JSONArray; +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.ChainSuspendException; +import com.easyagents.flow.core.chain.DataType; +import com.easyagents.flow.core.chain.EventManager; +import com.easyagents.flow.core.chain.Parameter; +import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository; +import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository; +import com.easyagents.flow.core.chain.runtime.Trigger; +import com.easyagents.flow.core.chain.runtime.TriggerContext; +import com.easyagents.flow.core.chain.runtime.TriggerType; +import com.easyagents.flow.core.node.ConfirmNode; +import com.easyagents.flow.core.parser.impl.ConfirmNodeParser; +import org.junit.Assert; +import org.junit.Test; + +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * 用户确认节点选择与输出契约测试。 + */ +public class ConfirmNodeTest { + + @Test + public void shouldBuildSingleChoiceAndReturnSelectedContent() { + ConfirmNode node = parse(false); + node.setId("confirm-1"); + Chain chain = createChain(); + + Parameter parameter = suspend(node, chain); + Assert.assertEquals("selection__confirm-1", parameter.getName()); + Assert.assertEquals("radio", parameter.getFormType()); + Assert.assertEquals("选择内容", parameter.getFormLabel()); + Assert.assertEquals(DataType.String, parameter.getDataType()); + Assert.assertEquals("第一议题", parameter.getOptions().get(0).getLabel()); + Assert.assertEquals("第一议题", parameter.getOptions().get(0).getValue()); + + chain.getExecutionState().getMemory().put(parameter.getName(), "审议类"); + Map result = executeAsResume(node, chain); + + Assert.assertEquals(Collections.singletonMap("selection", "审议类"), result); + Assert.assertFalse(chain.getExecutionState().getMemory() + .containsKey(parameter.getName())); + } + + @Test + public void shouldBuildMultipleChoiceAndReturnSelectedContents() { + ConfirmNode node = parse(true); + node.setId("confirm-1"); + Chain chain = createChain(); + + Parameter parameter = suspend(node, chain); + Assert.assertEquals("checkbox", parameter.getFormType()); + Assert.assertEquals(DataType.Array_String, parameter.getDataType()); + + List selected = List.of("第一议题", "听取类"); + chain.getExecutionState().getMemory().put(parameter.getName(), selected); + + Assert.assertEquals(selected, executeAsResume(node, chain).get("selection")); + } + + @Test + public void shouldUseConfiguredOutputNameWithoutChangingSuspendParameter() { + ConfirmNode node = parse(false, "templateChoice"); + node.setId("confirm-1"); + Chain chain = createChain(); + + Parameter parameter = suspend(node, chain); + Assert.assertEquals("selection__confirm-1", parameter.getName()); + + chain.getExecutionState().getMemory().put( + parameter.getName(), "第一议题"); + Assert.assertEquals( + Collections.singletonMap("templateChoice", "第一议题"), + executeAsResume(node, chain)); + } + + @Test + public void shouldIgnorePrefilledValueWithoutResumeTrigger() { + ConfirmNode node = parse(false); + node.setId("confirm-1"); + Chain chain = createChain(); + chain.getExecutionState().getMemory().put( + "selection__confirm-1", "未配置值"); + + Parameter parameter = suspend(node, chain); + + Assert.assertEquals("selection__confirm-1", parameter.getName()); + Assert.assertFalse(chain.getExecutionState().getMemory() + .containsKey(parameter.getName())); + } + + @Test + public void shouldRejectDuplicateNormalizedOptionContents() { + ConfirmNode node = parse(false); + node.setOptions(List.of("审议类", " 审议类 ")); + + try { + node.validateConfiguration(); + Assert.fail("duplicate option contents must be rejected"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("选项内容重复")); + } + } + + @Test + public void shouldRejectNonStringOptionDuringParsing() { + JSONObject data = data(false); + data.getJSONArray("options").add(1); + + try { + new ConfirmNodeParser().doParse( + new JSONObject(), data, new JSONObject()); + Assert.fail("non-string option must be rejected"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("必须为字符串")); + } + } + + @Test + public void shouldRejectStringEncodedOptionsDuringParsing() { + JSONObject data = data(false); + data.put("options", "[\"第一议题\"]"); + + try { + new ConfirmNodeParser().doParse( + new JSONObject(), data, new JSONObject()); + Assert.fail("string encoded options must be rejected"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("必须为数组")); + } + } + + @Test + public void shouldRejectImplicitModeConversionDuringParsing() { + JSONObject data = data(false); + data.put("multiple", "false"); + + try { + new ConfirmNodeParser().doParse( + new JSONObject(), data, new JSONObject()); + Assert.fail("non-boolean mode must be rejected"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("必须为布尔值")); + } + } + + @Test + public void shouldRejectUnknownConfigurationFieldDuringParsing() { + JSONObject data = data(false); + data.put("async", true); + + try { + new ConfirmNodeParser().doParse( + new JSONObject(), data, new JSONObject()); + Assert.fail("unknown confirm configuration must be rejected"); + } catch (IllegalArgumentException expected) { + Assert.assertTrue(expected.getMessage().contains("无效配置字段")); + } + } + + private static ConfirmNode parse(boolean multiple) { + return parse(multiple, ConfirmNode.DEFAULT_OUTPUT_NAME); + } + + private static ConfirmNode parse(boolean multiple, String outputName) { + ConfirmNode node = new ConfirmNodeParser().doParse( + new JSONObject(), data(multiple), new JSONObject()); + Parameter output = new Parameter(); + output.setName(outputName); + output.setDataType(multiple + ? DataType.Array_String + : DataType.String); + node.setOutputDefs(Collections.singletonList(output)); + node.validateConfiguration(); + return node; + } + + private static JSONObject data(boolean multiple) { + JSONObject data = new JSONObject(); + data.put("message", "请选择会议纪要模板"); + data.put("multiple", multiple); + JSONArray options = new JSONArray(); + options.addAll(List.of("第一议题", "审议类", "听取类")); + data.put("options", options); + return data; + } + + private static Parameter suspend(ConfirmNode node, Chain chain) { + try { + node.execute(chain); + throw new AssertionError("confirm node must suspend"); + } catch (ChainSuspendException expected) { + Assert.assertEquals(1, expected.getSuspendParameters().size()); + return expected.getSuspendParameters().get(0); + } + } + + private static Map executeAsResume( + ConfirmNode node, Chain chain) { + Trigger trigger = new Trigger(); + trigger.setType(TriggerType.RESUME); + trigger.setStateInstanceId(chain.getStateInstanceId()); + trigger.setNodeId(node.getId()); + TriggerContext.setCurrentTrigger(trigger); + try { + return node.execute(chain); + } finally { + TriggerContext.clearCurrentTrigger(); + } + } + + private static Chain createChain() { + ChainDefinition definition = new ChainDefinition(); + definition.setId("confirm-node-test"); + definition.setNodes(Collections.emptyList()); + definition.setEdges(Collections.emptyList()); + Chain chain = new Chain(definition, "confirm-node-instance"); + chain.setChainStateRepository(new InMemoryChainStateRepository()); + chain.setNodeStateRepository(new InMemoryNodeStateRepository()); + chain.setEventManager(new EventManager()); + return chain; + } +}