feat: 重构用户确认节点选择与恢复契约
- 统一单选多选输出与结构化暂停参数 - 增加严格恢复校验及并发状态保护 - 补充确认节点契约与恢复测试
This commit is contained in:
@@ -1751,16 +1751,91 @@ public class Chain {
|
|||||||
stateInstanceId,
|
stateInstanceId,
|
||||||
10L,
|
10L,
|
||||||
TimeUnit.SECONDS,
|
TimeUnit.SECONDS,
|
||||||
() -> {
|
() -> resumeSuspended(variables));
|
||||||
ChainState current =
|
|
||||||
chainStateRepository.load(stateInstanceId);
|
|
||||||
if (current == null
|
|
||||||
|| current.getStatus() != ChainStatus.SUSPEND) {
|
|
||||||
return false;
|
|
||||||
}
|
}
|
||||||
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;
|
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 恢复时注入的变量
|
* @param variables 恢复时注入的变量
|
||||||
*/
|
*/
|
||||||
private void resumeSuspended(Map<String, Object> variables) {
|
private boolean resumeSuspended(Map<String, Object> variables) {
|
||||||
ChainState newState = updateStateSafely(state -> {
|
AtomicBoolean resumed = new AtomicBoolean(false);
|
||||||
if (variables != null) {
|
AtomicReference<Set<String>> suspendedNodeIds =
|
||||||
state.getMemory().putAll(variables);
|
new AtomicReference<>(Collections.emptySet());
|
||||||
return EnumSet.of(ChainStateField.MEMORY);
|
updateStateSafely(state -> {
|
||||||
} else {
|
resumed.set(false);
|
||||||
|
suspendedNodeIds.set(Collections.emptySet());
|
||||||
|
if (state.getStatus() != ChainStatus.SUSPEND) {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
});
|
validateResumeVariables(state, variables);
|
||||||
|
if (state.getSuspendNodeIds() != null) {
|
||||||
notifyEvent(new ChainResumeEvent(this, variables));
|
suspendedNodeIds.set(
|
||||||
setStatusAndNotifyEvent(ChainStatus.RUNNING);
|
new LinkedHashSet<>(state.getSuspendNodeIds()));
|
||||||
|
}
|
||||||
Set<String> suspendNodeIds = newState.getSuspendNodeIds();
|
EnumSet<ChainStateField> updatedFields = EnumSet.of(
|
||||||
if (suspendNodeIds != null && !suspendNodeIds.isEmpty()) {
|
ChainStateField.STATUS,
|
||||||
// 移除 suspend 状态,方便二次 suspend 时,不带有旧数据
|
ChainStateField.SUSPEND_NODE_IDS,
|
||||||
updateStateSafely(state -> {
|
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.setSuspendNodeIds(null);
|
||||||
state.setSuspendForParameters(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);
|
Node node = definition.getNodeById(id);
|
||||||
if (node == null) {
|
if (node == null) {
|
||||||
throw new ChainException("Node not found: " + id);
|
throw new ChainException("Node not found: " + id);
|
||||||
}
|
}
|
||||||
scheduleNode(node, null, TriggerType.RESUME, 0L);
|
scheduleNode(node, null, TriggerType.RESUME, 0L);
|
||||||
}
|
}
|
||||||
}
|
return true;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void resume() {
|
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<Object> enums;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 显示文案与实际值分离的结构化选项。
|
||||||
|
*/
|
||||||
|
protected List<ParameterOption> options;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 用户输入的表单类型,例如:"input" "textarea" "select" "radio" "checkbox" 等等
|
* 用户输入的表单类型,例如:"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() {
|
public String getFormType() {
|
||||||
return formType;
|
return formType;
|
||||||
}
|
}
|
||||||
@@ -298,6 +311,7 @@ public class Parameter implements Serializable, Cloneable {
|
|||||||
", flattenAggregation=" + flattenAggregation +
|
", flattenAggregation=" + flattenAggregation +
|
||||||
", children=" + children +
|
", children=" + children +
|
||||||
", enums=" + enums +
|
", enums=" + enums +
|
||||||
|
", options=" + options +
|
||||||
", formType='" + formType + '\'' +
|
", formType='" + formType + '\'' +
|
||||||
", formLabel='" + formLabel + '\'' +
|
", formLabel='" + formLabel + '\'' +
|
||||||
", formPlaceholder='" + formPlaceholder + '\'' +
|
", formPlaceholder='" + formPlaceholder + '\'' +
|
||||||
@@ -320,6 +334,13 @@ public class Parameter implements Serializable, Cloneable {
|
|||||||
clone.enums = new ArrayList<>(this.enums.size());
|
clone.enums = new ArrayList<>(this.enums.size());
|
||||||
clone.enums.addAll(this.enums);
|
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;
|
return clone;
|
||||||
} catch (CloneNotSupportedException e) {
|
} catch (CloneNotSupportedException e) {
|
||||||
throw new AssertionError();
|
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;
|
package com.easyagents.flow.core.node;
|
||||||
|
|
||||||
|
|
||||||
import com.easyagents.flow.core.chain.Chain;
|
import com.easyagents.flow.core.chain.Chain;
|
||||||
import com.easyagents.flow.core.chain.ChainSuspendException;
|
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.Parameter;
|
||||||
|
import com.easyagents.flow.core.chain.ParameterOption;
|
||||||
import com.easyagents.flow.core.chain.RefType;
|
import com.easyagents.flow.core.chain.RefType;
|
||||||
import com.easyagents.flow.core.chain.repository.ChainStateField;
|
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 {
|
public class ConfirmNode extends BaseNode {
|
||||||
private static final long serialVersionUID = 1L;
|
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 String message;
|
||||||
private List<Parameter> confirms;
|
private boolean multiple;
|
||||||
|
private List<String> options;
|
||||||
|
|
||||||
public String getMessage() {
|
public String getMessage() {
|
||||||
return message;
|
return message;
|
||||||
@@ -39,115 +73,152 @@ public class ConfirmNode extends BaseNode {
|
|||||||
this.message = message;
|
this.message = message;
|
||||||
}
|
}
|
||||||
|
|
||||||
public List<Parameter> getConfirms() {
|
public boolean isMultiple() {
|
||||||
return confirms;
|
return multiple;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setConfirms(List<Parameter> confirms) {
|
public void setMultiple(boolean multiple) {
|
||||||
if (confirms != null) {
|
this.multiple = multiple;
|
||||||
for (Parameter confirm : confirms) {
|
|
||||||
confirm.setRefType(RefType.INPUT);
|
|
||||||
confirm.setRequired(true); // 必填,才能正确通过 getParameterValuesOnly 获取参数值
|
|
||||||
confirm.setName(confirm.getName());
|
|
||||||
}
|
|
||||||
}
|
|
||||||
this.confirms = confirms;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public List<String> getOptions() {
|
||||||
|
return options;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void setOptions(List<String> options) {
|
||||||
|
this.options = options;
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public Map<String, Object> execute(Chain chain) {
|
public Map<String, Object> execute(Chain chain) {
|
||||||
|
validateConfiguration();
|
||||||
|
String outputName = resolveOutputName();
|
||||||
|
Parameter parameter = buildParameter();
|
||||||
|
|
||||||
List<Parameter> confirmParameters = new ArrayList<>();
|
// 确认值只能来自经过 Chain.resumeIfSuspended 校验后创建的恢复触发器。
|
||||||
addConfirmParameter(confirmParameters);
|
// 启动参数与普通节点内存中的同名值均不能绕过人工确认。
|
||||||
|
if (!isValidatedResumeTrigger(chain)) {
|
||||||
if (confirms != null) {
|
chain.updateStateSafely(state -> {
|
||||||
for (Parameter confirm : confirms) {
|
if (state.getMemory().remove(parameter.getName()) == null) {
|
||||||
Parameter clone = confirm.clone();
|
return null;
|
||||||
clone.setName(confirm.getName() + "__" + getId());
|
|
||||||
clone.setRefType(RefType.INPUT);
|
|
||||||
confirmParameters.add(clone);
|
|
||||||
}
|
}
|
||||||
|
return EnumSet.of(ChainStateField.MEMORY);
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
Map<String, Object> values;
|
Map<String, Object> values;
|
||||||
try {
|
try {
|
||||||
values = chain.getExecutionState()
|
values = chain.getExecutionState()
|
||||||
.resolveParameters(this, confirmParameters);
|
.resolveParameters(this, Collections.singletonList(parameter));
|
||||||
// 移除 confirm 参数,方便在其他节点二次确认,或者在 for 循环中第二次获取
|
|
||||||
chain.updateStateSafely(state -> {
|
chain.updateStateSafely(state -> {
|
||||||
for (Parameter confirmParameter : confirmParameters) {
|
if (!state.getMemory().containsKey(parameter.getName())) {
|
||||||
state.getMemory().remove(confirmParameter.getName());
|
return null;
|
||||||
}
|
}
|
||||||
|
state.getMemory().remove(parameter.getName());
|
||||||
return EnumSet.of(ChainStateField.MEMORY);
|
return EnumSet.of(ChainStateField.MEMORY);
|
||||||
});
|
});
|
||||||
} catch (ChainSuspendException e) {
|
} catch (ChainSuspendException exception) {
|
||||||
chain.updateStateSafely(state -> {
|
chain.updateStateSafely(state -> {
|
||||||
state.setMessage(message);
|
state.setMessage(message);
|
||||||
return EnumSet.of(ChainStateField.MESSAGE);
|
return EnumSet.of(ChainStateField.MESSAGE);
|
||||||
});
|
});
|
||||||
|
throw exception;
|
||||||
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);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 获取参数值,不会触发 ChainSuspendException 错误
|
return Collections.singletonMap(
|
||||||
Map<String, Object> parameterValues =
|
outputName,
|
||||||
chain.getExecutionState().resolveParameters(
|
values.get(parameter.getName()));
|
||||||
this,
|
}
|
||||||
newParameters,
|
|
||||||
null,
|
|
||||||
true);
|
|
||||||
|
|
||||||
// 设置 enums,方便前端给用户进行选择
|
/**
|
||||||
for (Parameter confirmParameter : confirmParameters) {
|
* 获取并校验当前节点配置的唯一输出名称。
|
||||||
if (confirmParameter.getEnums() == null) {
|
*
|
||||||
Object enumsObject = parameterValues.get(confirmParameter.getName());
|
* @return 用户配置的输出名称
|
||||||
confirmParameter.setEnumsObject(enumsObject);
|
*/
|
||||||
|
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;
|
private Parameter buildParameter() {
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
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) {
|
|
||||||
// “确认 和 取消” 的参数
|
|
||||||
Parameter parameter = new Parameter();
|
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.setRefType(RefType.INPUT);
|
||||||
parameter.setId("confirm");
|
|
||||||
parameter.setName("confirm__" + getId());
|
|
||||||
parameter.setRequired(true);
|
parameter.setRequired(true);
|
||||||
|
|
||||||
List<Object> selectionData = new ArrayList<>();
|
|
||||||
selectionData.add("yes");
|
|
||||||
selectionData.add("no");
|
|
||||||
|
|
||||||
parameter.setEnums(selectionData);
|
|
||||||
parameter.setContentType("text");
|
parameter.setContentType("text");
|
||||||
parameter.setFormType("confirm");
|
parameter.setFormType(multiple ? "checkbox" : "radio");
|
||||||
parameters.add(parameter);
|
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;
|
package com.easyagents.flow.core.parser.impl;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson.JSONArray;
|
||||||
import com.alibaba.fastjson.JSONObject;
|
import com.alibaba.fastjson.JSONObject;
|
||||||
import com.easyagents.flow.core.chain.Parameter;
|
|
||||||
import com.easyagents.flow.core.node.ConfirmNode;
|
import com.easyagents.flow.core.node.ConfirmNode;
|
||||||
import com.easyagents.flow.core.parser.BaseNodeParser;
|
import com.easyagents.flow.core.parser.BaseNodeParser;
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
public class ConfirmNodeParser extends BaseNodeParser<ConfirmNode> {
|
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) {
|
public ConfirmNode doParse(JSONObject root, JSONObject data, JSONObject chainJSONObject) {
|
||||||
|
|
||||||
ConfirmNode confirmNode = new ConfirmNode();
|
ConfirmNode confirmNode = new ConfirmNode();
|
||||||
confirmNode.setMessage(data.getString("message"));
|
for (String key : data.keySet()) {
|
||||||
|
if (!ConfirmNode.SUPPORTED_CONFIGURATION_KEYS.contains(key)) {
|
||||||
List<Parameter> confirms = getParameters(data, "confirms");
|
throw new IllegalArgumentException(
|
||||||
if (confirms != null && !confirms.isEmpty()) {
|
"用户确认节点包含无效配置字段: " + key);
|
||||||
confirmNode.setConfirms(confirms);
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
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;
|
return confirmNode;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -15,10 +15,15 @@
|
|||||||
*/
|
*/
|
||||||
package com.easyagents.flow.core.test;
|
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.Chain;
|
||||||
import com.easyagents.flow.core.chain.ChainDefinition;
|
import com.easyagents.flow.core.chain.ChainDefinition;
|
||||||
import com.easyagents.flow.core.chain.ChainStatus;
|
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.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.ChainState;
|
||||||
import com.easyagents.flow.core.chain.event.ChainEndEvent;
|
import com.easyagents.flow.core.chain.event.ChainEndEvent;
|
||||||
import com.easyagents.flow.core.chain.repository.ChainDefinitionSnapshotRepository;
|
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.BaseNode;
|
||||||
import com.easyagents.flow.core.node.ConfirmNode;
|
import com.easyagents.flow.core.node.ConfirmNode;
|
||||||
import com.easyagents.flow.core.node.StartNode;
|
import com.easyagents.flow.core.node.StartNode;
|
||||||
|
import com.easyagents.flow.core.parser.ChainParser;
|
||||||
import org.junit.Assert;
|
import org.junit.Assert;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
||||||
@@ -61,6 +67,127 @@ import java.util.concurrent.atomic.AtomicReference;
|
|||||||
*/
|
*/
|
||||||
public class ChainExecutorConcurrencyTest {
|
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 入口遇到人工挂起会快速失败,不会无限占用调用线程。
|
* 验证同步 Tool 入口遇到人工挂起会快速失败,不会无限占用调用线程。
|
||||||
*
|
*
|
||||||
@@ -563,6 +690,12 @@ public class ChainExecutorConcurrencyTest {
|
|||||||
start.setId("start");
|
start.setId("start");
|
||||||
ConfirmNode confirm = new ConfirmNode();
|
ConfirmNode confirm = new ConfirmNode();
|
||||||
confirm.setId("confirm");
|
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();
|
EndNode end = new EndNode();
|
||||||
end.setId("end");
|
end.setId("end");
|
||||||
Edge first = new Edge();
|
Edge first = new Edge();
|
||||||
@@ -581,6 +714,91 @@ public class ChainExecutorConcurrencyTest {
|
|||||||
return definition;
|
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.Chain;
|
||||||
import com.easyagents.flow.core.chain.ChainDefinition;
|
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.ChainStatus;
|
||||||
import com.easyagents.flow.core.chain.EventManager;
|
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.InMemoryChainStateRepository;
|
||||||
import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
|
import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
|
||||||
import org.junit.Assert;
|
import org.junit.Assert;
|
||||||
import org.junit.Test;
|
import org.junit.Test;
|
||||||
|
|
||||||
|
import java.util.Arrays;
|
||||||
import java.util.Collections;
|
import java.util.Collections;
|
||||||
|
import java.util.EnumSet;
|
||||||
|
import java.util.List;
|
||||||
import java.util.Map;
|
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} 暂停恢复状态守卫测试。
|
* {@link Chain} 暂停恢复状态守卫测试。
|
||||||
@@ -72,6 +86,167 @@ public class ChainResumeGuardTest {
|
|||||||
.containsKey("unexpected"));
|
.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());
|
chain.setEventManager(new EventManager());
|
||||||
return chain;
|
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