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

发布 v1.10 #5

Merged
czm merged 147 commits from develop into main 2026-08-20 11:36:27 +08:00
Conversation 0 Commits 147 Files Changed 1488 +296 -7
10 changed files with 296 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4a8e633083 - Show all commits

7
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandAction.java
View File

@@ -13,5 +13,10 @@ public enum AgentRuntimeCommandAction {
/**
* 拒绝工具执行。
*/
REJECT
REJECT,
/**
* 审批过期并取消工具执行。
*/
EXPIRE
}

3
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumer.java
View File

@@ -90,6 +90,9 @@ public class AgentRuntimeCommandConsumer implements MQConsumerHandler {
agentRunService.rejectRuntimeLocal(
command.getRequestId(), command.getResumeToken(), command.getReason(),
command.getOperatorId(), command.getUserId());
} else if (command.getAction() == AgentRuntimeCommandAction.EXPIRE) {
agentRunService.expireApprovalLocal(
command.getRequestId(), command.getResumeToken(), command.getReason());
} else {
markFailureQuietly(command, new IllegalArgumentException("不支持的 Agent 远程运行命令"));
LOG.warn("跳过不支持的 Agent 远程运行命令: messageId={}, commandId={}, action={}",

15
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandProducer.java
View File

@@ -91,6 +91,21 @@ public class AgentRuntimeCommandProducer {
sendAndWait(targetNodeId, requestId, resumeToken, AgentRuntimeCommandAction.REJECT, reason, operatorId, userId);
}
/**
* 投递远程审批过期命令。
*
* @param targetNodeId 目标节点 ID
* @param requestId 请求 ID
* @param resumeToken 恢复令牌
* @param reason 过期原因
*/
public void sendExpire(String targetNodeId,
String requestId,
String resumeToken,
String reason) {
sendAndWait(targetNodeId, requestId, resumeToken, AgentRuntimeCommandAction.EXPIRE, reason, null, null);
}
private void sendAndWait(String targetNodeId,
String requestId,
String resumeToken,

44
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java
View File

@@ -74,6 +74,7 @@ public class AgentRunService {
private static final Logger LOG = LoggerFactory.getLogger(AgentRunService.class);
private static final String ASSISTANT_CODE = "AGENT";
private static final String DRAFT_ASSISTANT_CODE = "AGENT_DRAFT";
private static final String HITL_APPROVAL_EXPIRED_REASON = "审批超时，已自动拒绝";
@Resource
private AgentService agentService;
@@ -338,6 +339,38 @@ public class AgentRunService {
() -> agentHitlPendingService.reject(resumeToken, operatorId, reason));
}
/**
* 将已经持久化为过期状态的审批同步到运行节点。
*
* @param requestId 请求 ID
* @param resumeToken 恢复令牌
*/
public void expireApproval(String requestId, String resumeToken) {
if (agentRunRegistry.containsResumeTarget(requestId, resumeToken)) {
expireApprovalLocal(requestId, resumeToken, HITL_APPROVAL_EXPIRED_REASON);
return;
}
dispatchRemoteRuntimeCommand(requestId, resumeToken, AgentRuntimeCommandAction.EXPIRE,
HITL_APPROVAL_EXPIRED_REASON, null, null);
}
/**
* 在当前运行节点取消已经过期的审批。
*
* <p>数据库记录已由过期任务原子更新，本方法只恢复 runtime 的拒绝分支，
* 避免再次消费持久化 pending。</p>
*
* @param requestId 请求 ID
* @param resumeToken 恢复令牌
* @param reason 过期原因
*/
public void expireApprovalLocal(String requestId, String resumeToken, String reason) {
String resolvedReason = reason == null || reason.isBlank()
? HITL_APPROVAL_EXPIRED_REASON
: reason;
agentRunRegistry.reject(requestId, resumeToken, null, resolvedReason);
}
private void dispatchRemoteRuntimeCommand(String requestId,
String resumeToken,
AgentRuntimeCommandAction action,
@@ -364,7 +397,16 @@ public class AgentRunService {
agentRuntimeCommandProducer.sendApprove(ownerNodeId, resolvedRequestId, resumeToken, operatorId, userId);
return;
}
agentRuntimeCommandProducer.sendReject(ownerNodeId, resolvedRequestId, resumeToken, reason, operatorId, userId);
if (action == AgentRuntimeCommandAction.REJECT) {
agentRuntimeCommandProducer.sendReject(
ownerNodeId, resolvedRequestId, resumeToken, reason, operatorId, userId);
return;
}
if (action == AgentRuntimeCommandAction.EXPIRE) {
agentRuntimeCommandProducer.sendExpire(ownerNodeId, resolvedRequestId, resumeToken, reason);
return;
}
throw new BusinessException("不支持的 Agent 运行命令");
}
private String resolveRequestIdForRemoteDispatch(String requestId, String resumeToken) {

28
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTask.java
View File

@@ -5,6 +5,7 @@ import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import tech.easyflow.agent.entity.AgentHitlPending;
import tech.easyflow.agent.runtime.AgentRunService;
import tech.easyflow.common.cache.DistributedScheduledLock;
import java.util.List;
@@ -19,14 +20,18 @@ public class AgentHitlPendingExpirationTask {
private static final int BATCH_SIZE = 100;
private final AgentHitlPendingService pendingService;
private final AgentRunService agentRunService;
/**
* 创建任务。
*
* @param pendingService pending 服务
* @param agentRunService Agent 运行服务
*/
public AgentHitlPendingExpirationTask(AgentHitlPendingService pendingService) {
public AgentHitlPendingExpirationTask(AgentHitlPendingService pendingService,
AgentRunService agentRunService) {
this.pendingService = pendingService;
this.agentRunService = agentRunService;
}
/**
@@ -40,8 +45,29 @@ public class AgentHitlPendingExpirationTask {
if (!expired.isEmpty()) {
LOG.info("Expired Agent HITL pending records, count={}", expired.size());
}
for (AgentHitlPending pending : expired) {
notifyRuntimeExpired(pending);
}
} catch (RuntimeException e) {
LOG.warn("Expire Agent HITL pending records failed, message={}", e.getMessage(), e);
}
}
/**
* 通知审批所属运行节点进入过期取消分支。
*
* @param pending 已过期审批记录
*/
private void notifyRuntimeExpired(AgentHitlPending pending) {
if (pending == null) {
return;
}
try {
agentRunService.expireApproval(pending.getRequestId(), pending.getResumeToken());
} catch (RuntimeException e) {
// 单条运行态已结束或远程节点不可用时继续处理本批其他过期记录。
LOG.warn("Notify expired Agent HITL runtime failed, requestId={}, message={}",
pending.getRequestId(), e.getMessage(), e);
}
}
}

23
easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImpl.java
View File

@@ -13,6 +13,7 @@ import tech.easyflow.core.runtime.ChatRuntimeContext;
import java.math.BigInteger;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
@@ -147,14 +148,28 @@ public class AgentHitlPendingServiceImpl implements AgentHitlPendingService {
.le("expires_at", new Date())
.limit(Math.max(1, limit)));
Date now = new Date();
List<AgentHitlPending> expired = new ArrayList<>(records.size());
for (AgentHitlPending record : records) {
record.setStatus(AgentHitlPendingStatus.EXPIRED.name());
record.setRejectReason("审批超时，已自动拒绝");
AgentHitlPending update = new AgentHitlPending();
update.setStatus(AgentHitlPendingStatus.EXPIRED.name());
update.setRejectReason("审批超时，已自动拒绝");
update.setConsumedAt(now);
update.setModified(now);
// 用 status=PENDING 作为过期条件，避免定时任务覆盖刚刚完成的人工审批。
int updated = pendingMapper.updateByQuery(update, QueryWrapper.create()
.eq("id", record.getId())
.eq("status", AgentHitlPendingStatus.PENDING.name())
.eq("is_deleted", 0));
if (updated <= 0) {
continue;
}
record.setStatus(update.getStatus());
record.setRejectReason(update.getRejectReason());
record.setConsumedAt(now);
record.setModified(now);
pendingMapper.update(record);
expired.add(record);
}
return records;
return expired;
}
private AgentHitlPending consume(String resumeToken,

35
easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumerTest.java
View File

@@ -87,6 +87,32 @@ public class AgentRuntimeCommandConsumerTest {
Assert.assertNull(resultRegistry.lastFailureCommandId);
}
/**
* 验证审批过期命令走专用运行时入口，不重复消费持久化 pending。
*
* @throws Exception 消息序列化异常
*/
@Test
public void consumerShouldHandleExpireCommand() throws Exception {
AgentRuntimeProperties properties = new AgentRuntimeProperties();
properties.setInstanceId("node-a");
MQProperties mqProperties = new MQProperties();
RecordingAgentRunService service = new RecordingAgentRunService();
RecordingCommandResultRegistry resultRegistry = new RecordingCommandResultRegistry();
AgentRuntimeCommandConsumer consumer =
new AgentRuntimeCommandConsumer(new ObjectMapper(), properties, mqProperties, service, resultRegistry);
AgentRuntimeCommandMessage command = command("cmd-expire", "node-a");
command.setAction(AgentRuntimeCommandAction.EXPIRE);
command.setReason("expired");
consumer.handle(List.of(message(command)));
Assert.assertEquals(1, service.expireCount);
Assert.assertEquals("request-cmd-expire", service.lastRequestId);
Assert.assertEquals("expired", service.lastReason);
Assert.assertEquals("cmd-expire", resultRegistry.lastSuccessCommandId);
}
private AgentRuntimeCommandMessage command(String commandId, String targetNodeId) {
AgentRuntimeCommandMessage command = new AgentRuntimeCommandMessage();
command.setCommandId(commandId);
@@ -109,13 +135,22 @@ public class AgentRuntimeCommandConsumerTest {
private static final class RecordingAgentRunService extends AgentRunService {
private int approveCount;
private int expireCount;
private String lastRequestId;
private String lastReason;
@Override
public void approveRuntimeLocal(String requestId, String resumeToken, BigInteger operatorId, String userId) {
approveCount++;
lastRequestId = requestId;
}
@Override
public void expireApprovalLocal(String requestId, String resumeToken, String reason) {
expireCount++;
lastRequestId = requestId;
lastReason = reason;
}
}
private static class RecordingCommandResultRegistry extends AgentRuntimeCommandResultRegistry {

36
easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java
View File

@@ -685,6 +685,31 @@ public class AgentRunServiceDraftAndHitlTest {
Assert.assertEquals("request-remote-approve", commandProducer.lastRequestId);
}
/**
* 验证审批过期时会向远程 owner 投递专用命令。
*
* @throws Exception 运行态处理失败时抛出
*/
@Test
public void expireApprovalShouldDispatchRemoteWhenOwnerIsRemoteNode() throws Exception {
AgentRunService service = new AgentRunService();
RecordingRouteRegistry routeRegistry = new RecordingRouteRegistry("node-b");
routeRegistry.ownerNode = "node-a";
routeRegistry.ownerBootId = "boot-a";
routeRegistry.currentOwnerBootId = "boot-a";
routeRegistry.nodeAlive = true;
RecordingCommandProducer commandProducer = new RecordingCommandProducer();
setField(service, "agentRunRegistry", new AgentRunRegistry());
setField(service, "agentRuntimeRouteRegistry", routeRegistry);
setField(service, "agentRuntimeCommandProducer", commandProducer);
service.expireApproval("request-remote-expire", "token-remote-expire");
Assert.assertEquals(1, commandProducer.expireCount);
Assert.assertEquals("node-a", commandProducer.lastTargetNodeId);
Assert.assertEquals("request-remote-expire", commandProducer.lastRequestId);
}
/**
* 验证 owner 缺失时明确失败。
*
@@ -1073,6 +1098,7 @@ public class AgentRunServiceDraftAndHitlTest {
private static class RecordingCommandProducer extends AgentRuntimeCommandProducer {
private int approveCount;
private int expireCount;
private String lastTargetNodeId;
private String lastRequestId;
@@ -1086,6 +1112,16 @@ public class AgentRunServiceDraftAndHitlTest {
lastTargetNodeId = targetNodeId;
lastRequestId = requestId;
}
@Override
public void sendExpire(String targetNodeId,
String requestId,
String resumeToken,
String reason) {
expireCount++;
lastTargetNodeId = targetNodeId;
lastRequestId = requestId;
}
}
private static class RecordingAgentRuntimeFactory implements AgentRuntimeFactory {

50
easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTaskTest.java Normal file
View File

@@ -0,0 +1,50 @@
package tech.easyflow.agent.runtime.hitl;
import org.junit.Test;
import org.mockito.Mockito;
import tech.easyflow.agent.entity.AgentHitlPending;
import tech.easyflow.agent.runtime.AgentRunService;
import java.util.List;
/**
* {@link AgentHitlPendingExpirationTask} 回归测试。
*/
public class AgentHitlPendingExpirationTaskTest {
/**
* 验证数据库审批过期后通知所有运行节点，单条通知失败不阻断同批其他记录。
*/
@Test
public void expirePendingShouldNotifyEveryRuntime() {
AgentHitlPendingService pendingService = Mockito.mock(AgentHitlPendingService.class);
AgentRunService runService = Mockito.mock(AgentRunService.class);
AgentHitlPending first = pending("request-1", "token-1");
AgentHitlPending second = pending("request-2", "token-2");
Mockito.when(pendingService.expirePending(100)).thenReturn(List.of(first, second));
Mockito.doThrow(new RuntimeException("owner unavailable"))
.when(runService).expireApproval("request-1", "token-1");
AgentHitlPendingExpirationTask task =
new AgentHitlPendingExpirationTask(pendingService, runService);
task.expirePending();
Mockito.verify(runService).expireApproval("request-1", "token-1");
Mockito.verify(runService).expireApproval("request-2", "token-2");
}
/**
* 创建已过期审批记录。
*
* @param requestId 请求 ID
* @param resumeToken 恢复令牌
* @return 审批记录
*/
private AgentHitlPending pending(String requestId, String resumeToken) {
AgentHitlPending pending = new AgentHitlPending();
pending.setRequestId(requestId);
pending.setResumeToken(resumeToken);
pending.setStatus(AgentHitlPendingStatus.EXPIRED.name());
return pending;
}
}

62
easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImplTest.java Normal file
View File

@@ -0,0 +1,62 @@
package tech.easyflow.agent.runtime.hitl;
import com.mybatisflex.core.query.QueryWrapper;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import tech.easyflow.agent.config.AgentRuntimeProperties;
import tech.easyflow.agent.entity.AgentHitlPending;
import tech.easyflow.agent.mapper.AgentHitlPendingMapper;
import java.math.BigInteger;
import java.util.List;
/**
* {@link AgentHitlPendingServiceImpl} 回归测试。
*/
public class AgentHitlPendingServiceImplTest {
/**
* 验证过期扫描只返回成功从 PENDING 原子更新为 EXPIRED 的记录。
*/
@Test
public void expirePendingShouldIgnoreConcurrentlyConsumedRecord() {
AgentHitlPendingMapper mapper = Mockito.mock(AgentHitlPendingMapper.class);
AgentHitlPending first = pending(BigInteger.ONE, "request-1", "token-1");
AgentHitlPending concurrentlyApproved = pending(BigInteger.TWO, "request-2", "token-2");
Mockito.when(mapper.selectListByQuery(Mockito.any(QueryWrapper.class)))
.thenReturn(List.of(first, concurrentlyApproved));
Mockito.when(mapper.updateByQuery(
Mockito.any(AgentHitlPending.class), Mockito.any(QueryWrapper.class)))
.thenReturn(1, 0);
AgentHitlPendingServiceImpl service =
new AgentHitlPendingServiceImpl(mapper, new AgentRuntimeProperties());
List<AgentHitlPending> expired = service.expirePending(100);
Assert.assertEquals(1, expired.size());
Assert.assertSame(first, expired.get(0));
Assert.assertEquals(AgentHitlPendingStatus.EXPIRED.name(), first.getStatus());
Assert.assertEquals(AgentHitlPendingStatus.PENDING.name(), concurrentlyApproved.getStatus());
Mockito.verify(mapper, Mockito.times(2)).updateByQuery(
Mockito.any(AgentHitlPending.class), Mockito.any(QueryWrapper.class));
}
/**
* 创建待过期审批记录。
*
* @param id 记录 ID
* @param requestId 请求 ID
* @param resumeToken 恢复令牌
* @return 审批记录
*/
private AgentHitlPending pending(BigInteger id, String requestId, String resumeToken) {
AgentHitlPending pending = new AgentHitlPending();
pending.setId(id);
pending.setRequestId(requestId);
pending.setResumeToken(resumeToken);
pending.setStatus(AgentHitlPendingStatus.PENDING.name());
pending.setIsDeleted(0);
return pending;
}
}