EasyFlow-Tech-Dev/Easy-Agents
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完善工作流实例状态查询与恢复

Browse Source 
- 增加暂停态原子恢复守卫，避免重复恢复覆盖实例状态

- 支持从实例定义快照读取节点名称
This commit is contained in:
陈子默
2026-08-09 21:24:40 +08:00
parent f13e24751a
commit 8d8d77ffda
4 changed files with 280 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1732,7 +1732,47 @@ public class Chain {
}
/**
* 仅在工作流处于暂停状态时恢复执行。
*
* <p>状态判断与恢复动作在同一个实例锁内完成，避免并发恢复请求重复注入变量
* 或把终态实例重新改为运行中。</p>
*
* @param variables 恢复时注入的变量
* @return 本次是否完成了暂停态到运行态的转换
*/
public boolean resumeIfSuspended(Map<String, Object> variables) {
return executeWithLock(
stateInstanceId,
10L,
TimeUnit.SECONDS,
() -> {
ChainState current =
chainStateRepository.load(stateInstanceId);
if (current == null
|| current.getStatus() != ChainStatus.SUSPEND) {
return false;
}
resumeSuspended(variables);
return true;
});
}
/**
* 恢复暂停中的工作流。
*
* @param variables 恢复时注入的变量
*/
public void resume(Map<String, Object> variables) {
resumeIfSuspended(variables);
}
/**
* 在调用方持有实例锁且已确认暂停状态后执行恢复动作。
*
* @param variables 恢复时注入的变量
*/
private void resumeSuspended(Map<String, Object> variables) {
ChainState newState = updateStateSafely(state -> {
if (variables != null) {
state.getMemory().putAll(variables);

76
easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
View File

@@ -896,6 +896,82 @@ public class ChainExecutor {
chain.resume(variables);
}
/**
* 仅在工作流实例处于暂停状态时恢复执行。
*
* <p>状态判断和恢复由 {@link Chain} 在同一个实例锁内完成，可安全处理并发恢复请求。</p>
*
* @param stateInstanceId 工作流实例 ID
* @param variables 恢复时注入的变量
* @return 本次是否完成了暂停态到运行态的转换
*/
public boolean resumeAsyncIfSuspended(
String stateInstanceId,
Map<String, Object> variables) {
ChainState state = chainStateRepository.load(stateInstanceId);
if (state == null) {
return false;
}
ChainDefinition definition = getDefinitionForInstance(state);
if (definition == null) {
return false;
}
Chain chain = configureChain(
definition,
state.getInstanceId(),
state);
return chain.resumeIfSuspended(variables);
}
/**
* 获取工作流实例启动时定义快照中的节点名称。
*
* @param stateInstanceId 工作流实例 ID
* @return 按定义顺序排列的节点 ID 与名称；实例或定义不存在时返回空映射
*/
public Map<String, String> getInstanceNodeNames(
String stateInstanceId) {
ChainState state = chainStateRepository.load(stateInstanceId);
if (state == null) {
return Collections.emptyMap();
}
return getInstanceNodeNames(state);
}
/**
* 使用调用方已经加载的状态获取实例定义快照中的节点名称。
*
* @param state 已加载的工作流状态
* @return 按定义顺序排列的节点 ID 与名称；状态或定义不存在时返回空映射
*/
public Map<String, String> getInstanceNodeNames(
ChainState state) {
if (state == null) {
return Collections.emptyMap();
}
ChainDefinition definition = getDefinitionForInstance(state);
if (definition == null
|| definition.getNodes() == null
|| definition.getNodes().isEmpty()) {
return Collections.emptyMap();
}
Map<String, String> nodeNames = new LinkedHashMap<>();
for (Node node : definition.getNodes()) {
if (node == null
|| node.getId() == null
|| node.getId().isBlank()) {
continue;
}
String nodeName = node.getName();
nodeNames.put(
node.getId(),
nodeName == null || nodeName.isBlank()
? node.getId()
: nodeName);
}
return Collections.unmodifiableMap(nodeNames);
}
private Chain createChain(String definitionId) {
ChainDefinition definition = definitionRepository.getChainDefinitionById(definitionId);

68
easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorInstanceMetadataTest.java Normal file
View File

@@ -0,0 +1,68 @@
package com.easyagents.flow.core.test;
import com.easyagents.flow.core.chain.ChainDefinition;
import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository;
import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
import com.easyagents.flow.core.chain.runtime.ChainExecutor;
import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
import com.easyagents.flow.core.node.StartNode;
import org.junit.Assert;
import org.junit.Test;
import java.util.Collections;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
/**
* {@link ChainExecutor} 实例定义元数据查询测试。
*/
public class ChainExecutorInstanceMetadataTest {
/**
* 验证节点名称来自实例启动时可恢复的定义快照。
*/
@Test
public void shouldResolveNodeNamesFromInstanceDefinition() {
ChainDefinition definition = new ChainDefinition();
definition.setId("metadata-definition");
StartNode start = new StartNode();
start.setId("start");
start.setName("开始节点");
definition.setNodes(Collections.singletonList(start));
definition.setEdges(Collections.emptyList());
InMemoryChainStateRepository stateRepository =
new InMemoryChainStateRepository();
stateRepository.create("metadata-instance")
.setChainDefinitionId(definition.getId());
ScheduledExecutorService schedulerPool =
Executors.newSingleThreadScheduledExecutor();
ExecutorService workerPool =
Executors.newSingleThreadExecutor();
TriggerScheduler scheduler = new TriggerScheduler(
new InMemoryTriggerStore(),
schedulerPool,
workerPool,
1_000L);
ChainExecutor executor = new ChainExecutor(
ignored -> definition,
stateRepository,
new InMemoryNodeStateRepository(),
scheduler);
try {
Map<String, String> nodeNames =
executor.getInstanceNodeNames(
"metadata-instance");
Assert.assertEquals(
Map.of("start", "开始节点"),
nodeNames);
} finally {
scheduler.shutdown();
}
}
}

96
easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainResumeGuardTest.java Normal file
View File

@@ -0,0 +1,96 @@
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.ChainStatus;
import com.easyagents.flow.core.chain.EventManager;
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.Collections;
import java.util.Map;
/**
* {@link Chain} 暂停恢复状态守卫测试。
*/
public class ChainResumeGuardTest {
/**
* 验证只有暂停中的实例可以恢复，重复恢复不会再次注入变量。
*/
@Test
public void shouldResumeOnlyOnceFromSuspendedState() {
InMemoryChainStateRepository stateRepository =
new InMemoryChainStateRepository();
Chain chain = createChain(stateRepository, "resume-once");
chain.suspend();
boolean resumed = chain.resumeIfSuspended(
Map.of("approved", true));
boolean resumedAgain = chain.resumeIfSuspended(
Map.of("unexpected", true));
Assert.assertTrue(resumed);
Assert.assertFalse(resumedAgain);
Assert.assertEquals(
ChainStatus.RUNNING,
stateRepository.load("resume-once").getStatus());
Assert.assertEquals(
Boolean.TRUE,
stateRepository.load("resume-once")
.getMemory()
.get("approved"));
Assert.assertFalse(
stateRepository.load("resume-once")
.getMemory()
.containsKey("unexpected"));
}
/**
* 验证成功终态不会被恢复操作改回运行中。
*/
@Test
public void shouldKeepTerminalStateUnchanged() {
InMemoryChainStateRepository stateRepository =
new InMemoryChainStateRepository();
Chain chain = createChain(stateRepository, "resume-terminal");
stateRepository.load("resume-terminal")
.setStatus(ChainStatus.SUCCEEDED);
boolean resumed = chain.resumeIfSuspended(
Map.of("unexpected", true));
Assert.assertFalse(resumed);
Assert.assertEquals(
ChainStatus.SUCCEEDED,
stateRepository.load("resume-terminal").getStatus());
Assert.assertFalse(
stateRepository.load("resume-terminal")
.getMemory()
.containsKey("unexpected"));
}
/**
* 创建使用进程内状态仓储的最小工作流。
*
* @param stateRepository 状态仓储
* @param instanceId 工作流实例 ID
* @return 已配置工作流
*/
private Chain createChain(
InMemoryChainStateRepository stateRepository,
String instanceId) {
ChainDefinition definition = new ChainDefinition();
definition.setId("resume-definition");
definition.setNodes(Collections.emptyList());
definition.setEdges(Collections.emptyList());
Chain chain = new Chain(definition, instanceId);
chain.setChainStateRepository(stateRepository);
chain.setNodeStateRepository(
new InMemoryNodeStateRepository());
chain.setEventManager(new EventManager());
return chain;
}
}