diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandAction.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandAction.java
index 8758081c..763bc185 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandAction.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandAction.java
@@ -13,5 +13,10 @@ public enum AgentRuntimeCommandAction {
/**
* 拒绝工具执行。
*/
- REJECT
+ REJECT,
+
+ /**
+ * 审批过期并取消工具执行。
+ */
+ EXPIRE
}
diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumer.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumer.java
index 54acbdf9..02c484cb 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumer.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumer.java
@@ -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={}",
diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandProducer.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandProducer.java
index cff2847f..5d93197a 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandProducer.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/distributed/AgentRuntimeCommandProducer.java
@@ -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,
diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java
index cdebbcb4..08e18a65 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/AgentRunService.java
@@ -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);
+ }
+
+ /**
+ * 在当前运行节点取消已经过期的审批。
+ *
+ * 数据库记录已由过期任务原子更新，本方法只恢复 runtime 的拒绝分支，
+ * 避免再次消费持久化 pending。
+ *
+ * @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) {
diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTask.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTask.java
index f9bd2a3a..d4fc3db3 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTask.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTask.java
@@ -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);
+ }
+ }
}
diff --git a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImpl.java b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImpl.java
index e2c7d486..ee315f78 100644
--- a/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImpl.java
+++ b/easyflow-modules/easyflow-module-agent/src/main/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImpl.java
@@ -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 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,
diff --git a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumerTest.java b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumerTest.java
index 8faa291e..62f0ad76 100644
--- a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumerTest.java
+++ b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/distributed/AgentRuntimeCommandConsumerTest.java
@@ -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 {
diff --git a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java
index f259eb16..72f6afa7 100644
--- a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java
+++ b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/AgentRunServiceDraftAndHitlTest.java
@@ -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 {
diff --git a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTaskTest.java b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTaskTest.java
new file mode 100644
index 00000000..21ca6752
--- /dev/null
+++ b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingExpirationTaskTest.java
@@ -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;
+ }
+}
diff --git a/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImplTest.java b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImplTest.java
new file mode 100644
index 00000000..252b63b7
--- /dev/null
+++ b/easyflow-modules/easyflow-module-agent/src/test/java/tech/easyflow/agent/runtime/hitl/AgentHitlPendingServiceImplTest.java
@@ -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 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;
+ }
+}