feat: 重构用户确认节点选择与恢复契约
- 统一单选多选输出与结构化暂停参数 - 增加严格恢复校验及并发状态保护 - 补充确认节点契约与恢复测试
This commit is contained in:
@@ -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));
|
||||
}
|
||||
resumeSuspended(variables);
|
||||
|
||||
private void validateResumeVariables(
|
||||
ChainState state, Map<String, Object> variables) {
|
||||
List<Parameter> parameters = state.getSuspendForParameters();
|
||||
if (parameters == null || parameters.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Map<String, Object> submitted = variables == null
|
||||
? Collections.emptyMap()
|
||||
: variables;
|
||||
Set<String> expectedKeys = parameters.stream()
|
||||
.map(Parameter::getName)
|
||||
.filter(Objects::nonNull)
|
||||
.collect(java.util.stream.Collectors.toCollection(
|
||||
LinkedHashSet::new));
|
||||
Set<String> 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<ParameterOption> options = parameter.getOptions();
|
||||
if (options == null || options.isEmpty()) {
|
||||
return;
|
||||
}
|
||||
Set<String> 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<String> 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<String, Object> variables) {
|
||||
ChainState newState = updateStateSafely(state -> {
|
||||
if (variables != null) {
|
||||
state.getMemory().putAll(variables);
|
||||
return EnumSet.of(ChainStateField.MEMORY);
|
||||
} else {
|
||||
private boolean resumeSuspended(Map<String, Object> variables) {
|
||||
AtomicBoolean resumed = new AtomicBoolean(false);
|
||||
AtomicReference<Set<String>> suspendedNodeIds =
|
||||
new AtomicReference<>(Collections.emptySet());
|
||||
updateStateSafely(state -> {
|
||||
resumed.set(false);
|
||||
suspendedNodeIds.set(Collections.emptySet());
|
||||
if (state.getStatus() != ChainStatus.SUSPEND) {
|
||||
return null;
|
||||
}
|
||||
});
|
||||
|
||||
notifyEvent(new ChainResumeEvent(this, variables));
|
||||
setStatusAndNotifyEvent(ChainStatus.RUNNING);
|
||||
|
||||
Set<String> suspendNodeIds = newState.getSuspendNodeIds();
|
||||
if (suspendNodeIds != null && !suspendNodeIds.isEmpty()) {
|
||||
// 移除 suspend 状态,方便二次 suspend 时,不带有旧数据
|
||||
updateStateSafely(state -> {
|
||||
validateResumeVariables(state, variables);
|
||||
if (state.getSuspendNodeIds() != null) {
|
||||
suspendedNodeIds.set(
|
||||
new LinkedHashSet<>(state.getSuspendNodeIds()));
|
||||
}
|
||||
EnumSet<ChainStateField> 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);
|
||||
return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS, ChainStateField.SUSPEND_FOR_PARAMETERS);
|
||||
resumed.set(true);
|
||||
return updatedFields;
|
||||
});
|
||||
if (!resumed.get()) {
|
||||
return false;
|
||||
}
|
||||
|
||||
for (String id : suspendNodeIds) {
|
||||
notifyEvent(new ChainResumeEvent(this, variables));
|
||||
notifyEvent(new ChainStatusChangeEvent(
|
||||
this, ChainStatus.RUNNING, ChainStatus.SUSPEND));
|
||||
|
||||
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() {
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
/**
|
||||
* Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
|
||||
* <p>
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
@@ -49,6 +49,11 @@ public class Parameter implements Serializable, Cloneable {
|
||||
*/
|
||||
protected List<Object> enums;
|
||||
|
||||
/**
|
||||
* 显示文案与实际值分离的结构化选项。
|
||||
*/
|
||||
protected List<ParameterOption> options;
|
||||
|
||||
/**
|
||||
* 用户输入的表单类型,例如:"input" "textarea" "select" "radio" "checkbox" 等等
|
||||
*/
|
||||
@@ -242,6 +247,14 @@ public class Parameter implements Serializable, Cloneable {
|
||||
}
|
||||
}
|
||||
|
||||
public List<ParameterOption> getOptions() {
|
||||
return options;
|
||||
}
|
||||
|
||||
public void setOptions(List<ParameterOption> 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();
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
/**
|
||||
* Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
|
||||
* <p>
|
||||
* 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 + '\'' +
|
||||
'}';
|
||||
}
|
||||
}
|
||||
@@ -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<String> 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<Parameter> confirms;
|
||||
private boolean multiple;
|
||||
private List<String> options;
|
||||
|
||||
public String getMessage() {
|
||||
return message;
|
||||
@@ -39,115 +73,152 @@ public class ConfirmNode extends BaseNode {
|
||||
this.message = message;
|
||||
}
|
||||
|
||||
public List<Parameter> getConfirms() {
|
||||
return confirms;
|
||||
public boolean isMultiple() {
|
||||
return multiple;
|
||||
}
|
||||
|
||||
public void setConfirms(List<Parameter> 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<String> getOptions() {
|
||||
return options;
|
||||
}
|
||||
|
||||
public void setOptions(List<String> options) {
|
||||
this.options = options;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, Object> execute(Chain chain) {
|
||||
validateConfiguration();
|
||||
String outputName = resolveOutputName();
|
||||
Parameter parameter = buildParameter();
|
||||
|
||||
List<Parameter> 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<String, Object> 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<Parameter> newParameters = new ArrayList<>();
|
||||
for (Parameter confirm : confirms) {
|
||||
Parameter clone = confirm.clone();
|
||||
clone.setName(confirm.getName() + "__" + getId());
|
||||
clone.setRefType(RefType.REF); // 固定为 REF
|
||||
newParameters.add(clone);
|
||||
throw exception;
|
||||
}
|
||||
|
||||
// 获取参数值,不会触发 ChainSuspendException 错误
|
||||
Map<String, Object> parameterValues =
|
||||
chain.getExecutionState().resolveParameters(
|
||||
this,
|
||||
newParameters,
|
||||
null,
|
||||
true);
|
||||
return Collections.singletonMap(
|
||||
outputName,
|
||||
values.get(parameter.getName()));
|
||||
}
|
||||
|
||||
// 设置 enums,方便前端给用户进行选择
|
||||
for (Parameter confirmParameter : confirmParameters) {
|
||||
if (confirmParameter.getEnums() == null) {
|
||||
Object enumsObject = parameterValues.get(confirmParameter.getName());
|
||||
confirmParameter.setEnumsObject(enumsObject);
|
||||
/**
|
||||
* 获取并校验当前节点配置的唯一输出名称。
|
||||
*
|
||||
* @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<String> normalizedOptions = new HashSet<>();
|
||||
for (String option : options) {
|
||||
String normalized = requireText(option, MAX_OPTION_LENGTH, "选项内容");
|
||||
if (!normalizedOptions.add(normalized)) {
|
||||
throw new IllegalArgumentException("用户确认节点选项内容重复: " + normalized);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
throw e;
|
||||
}
|
||||
|
||||
|
||||
Map<String, Object> 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 results;
|
||||
}
|
||||
|
||||
|
||||
private void addConfirmParameter(List<Parameter> 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<Object> 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<ParameterOption> buildRuntimeOptions() {
|
||||
List<ParameterOption> 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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<ConfirmNode> {
|
||||
@@ -28,12 +29,36 @@ public class ConfirmNodeParser extends BaseNodeParser<ConfirmNode> {
|
||||
public ConfirmNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) {
|
||||
|
||||
ConfirmNode confirmNode = new ConfirmNode();
|
||||
confirmNode.setMessage(data.getString("message"));
|
||||
|
||||
List<Parameter> 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<String> 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;
|
||||
}
|
||||
|
||||
@@ -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<String> 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>" : "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;
|
||||
}
|
||||
|
||||
/**
|
||||
* 创建用于取消传播验证的工作流。
|
||||
*
|
||||
|
||||
@@ -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<Boolean> firstResult = executor.submit(() -> {
|
||||
ready.countDown();
|
||||
start.await();
|
||||
return first.resumeIfSuspended(
|
||||
Map.of("templateType__confirm", "AGENDA"));
|
||||
});
|
||||
Future<Boolean> 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<String, Object> 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<ChainStateField> fields) {
|
||||
updateCount.incrementAndGet();
|
||||
return super.tryUpdate(chainState, fields);
|
||||
}
|
||||
|
||||
private int getUpdateCount() {
|
||||
return updateCount.get();
|
||||
}
|
||||
|
||||
private void resetUpdateCount() {
|
||||
updateCount.set(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,235 @@
|
||||
/**
|
||||
* Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
|
||||
* <p>
|
||||
* 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<String, Object> 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<String> 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<String, Object> 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user