feat: 完善 Agent Skill 渐进披露运行时

- 支持 Skill 绑定 MCP 冻结清单和延迟注册 - 拒绝同步工具工作流进入不可恢复挂起状态
2026-08-19 21:51:16 +08:00
parent 49a7de34bb
commit c7d410d755
19 changed files with 1417 additions and 59 deletions
--- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
+++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java
@@ -194,18 +194,44 @@ public class ChainExecutor {


    public Map<String, Object> execute(String definitionId, Map<String, Object> variables) {
-        return execute(definitionId, variables, Long.MAX_VALUE, TimeUnit.SECONDS);
+        return executeInternal(definitionId, variables, Long.MAX_VALUE, TimeUnit.SECONDS, false);
    }


    public Map<String, Object> execute(String definitionId, Map<String, Object> variables, long timeout, TimeUnit unit) {
+        return executeInternal(definitionId, variables, timeout, unit, false);
+    }
+
+    /**
+     * 同步执行不允许进入人工挂起状态的工作流。
+     *
+     * <p>该入口适用于 Tool 等无法把工作流恢复协议接回原调用方的同步场景。
+     * 工作流一旦进入 {@link ChainStatus#SUSPEND}，实例会被取消并立即返回失败。</p>
+     *
+     * @param definitionId 工作流定义 ID
+     * @param variables 输入变量
+     * @return 工作流输出
+     * @throws RuntimeException 工作流失败、挂起或执行线程被中断时抛出
+     */
+    public Map<String, Object> executeWithoutSuspension(
+            String definitionId, Map<String, Object> variables) {
+        return executeInternal(
+                definitionId, variables, Long.MAX_VALUE, TimeUnit.SECONDS, true);
+    }
+
+    private Map<String, Object> executeInternal(
+            String definitionId,
+            Map<String, Object> variables,
+            long timeout,
+            TimeUnit unit,
+            boolean rejectSuspension) {
        Chain chain = createChain(definitionId);
        String stateInstanceId = chain.getStateInstanceId();

        try {
            chain.start(variables);
            Map<String, Object> result = awaitPersistentOutcome(
-                    stateInstanceId, timeout, unit, null);
+                    stateInstanceId, timeout, unit, null, rejectSuspension);
            clearDefaultStates(result);
            return result;
        } catch (TimeoutException e) {
@@ -759,6 +785,17 @@ public class ChainExecutor {
            TimeUnit unit,
            Chain parentChain)
            throws InterruptedException, TimeoutException {
+        return awaitPersistentOutcome(
+                stateInstanceId, timeout, unit, parentChain, false);
+    }
+
+    private Map<String, Object> awaitPersistentOutcome(
+            String stateInstanceId,
+            long timeout,
+            TimeUnit unit,
+            Chain parentChain,
+            boolean rejectSuspension)
+            throws InterruptedException, TimeoutException {
        Objects.requireNonNull(unit, "time unit required");
        long timeoutNanos = timeout == Long.MAX_VALUE
                ? Long.MAX_VALUE
@@ -789,6 +826,12 @@ public class ChainExecutor {
                        "Chain state not found: " + stateInstanceId);
            }
            ChainStatus status = state.getStatus();
+            if (rejectSuspension && status == ChainStatus.SUSPEND) {
+                cancel(stateInstanceId, "Suspended workflow is not supported by this caller");
+                throw new ChainException(
+                        "Workflow suspended and requires external input: "
+                                + stateInstanceId);
+            }
            if (status != null && status.isTerminal()) {
                if (!status.isSuccess()) {
                    ExceptionSummary error = state.getError();
@@ -1087,6 +1130,10 @@ public class ChainExecutor {
            // 状态已过期或被清理时，该触发器已经失去业务目标，直接确认避免无限热重放。
            return;
        }
+        if (state.getStatus() != null && state.getStatus().isTerminal()) {
+            // 终态不可再次执行；直接确认迟到或重复触发器，避免重新加载已清理的定义快照。
+            return;
+        }


        ChainDefinition definition = getDefinitionForInstance(state);
--- a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java
+++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java
@@ -33,6 +33,7 @@ import com.easyagents.flow.core.chain.runtime.Trigger;
 import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
 import com.easyagents.flow.core.node.EndNode;
 import com.easyagents.flow.core.node.BaseNode;
+import com.easyagents.flow.core.node.ConfirmNode;
 import com.easyagents.flow.core.node.StartNode;
 import org.junit.Assert;
 import org.junit.Test;
@@ -46,6 +47,7 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -59,6 +61,43 @@ import java.util.concurrent.atomic.AtomicReference;
 */
 public class ChainExecutorConcurrencyTest {

+    /**
+     * 验证同步 Tool 入口遇到人工挂起会快速失败，不会无限占用调用线程。
+     *
+     * @throws Exception 异步测试执行失败时抛出
+     */
+    @Test
+    public void shouldFailFastWhenNonSuspendingExecutionIsSuspended()
+            throws Exception {
+        ScheduledExecutorService schedulerPool = Executors.newSingleThreadScheduledExecutor();
+        ExecutorService workerPool = Executors.newFixedThreadPool(2);
+        TriggerScheduler triggerScheduler = new TriggerScheduler(
+                new InMemoryTriggerStore(), schedulerPool, workerPool, 10L);
+        ChainDefinition definition = createConfirmDefinition();
+        ChainExecutor executor = new ChainExecutor(
+                ignored -> definition,
+                new InMemoryChainStateRepository(),
+                new InMemoryNodeStateRepository(),
+                triggerScheduler);
+        ExecutorService caller = Executors.newSingleThreadExecutor();
+        try {
+            Future<Map<String, Object>> result = caller.submit(
+                    () -> executor.executeWithoutSuspension(
+                            definition.getId(), Collections.emptyMap()));
+            try {
+                result.get(3, TimeUnit.SECONDS);
+                Assert.fail("suspended workflow must fail");
+            } catch (ExecutionException exception) {
+                Assert.assertTrue(
+                        String.valueOf(exception.getCause().getMessage())
+                                .contains("Execution failed"));
+            }
+        } finally {
+            caller.shutdownNow();
+            triggerScheduler.shutdown();
+        }
+    }
+
    /**
     * 验证实例初始化会在入口触发器创建前持久化工作流定义 ID。
     *
@@ -512,6 +551,36 @@ public class ChainExecutorConcurrencyTest {
        return definition;
    }

+    /**
+     * 创建包含内部确认节点的测试 Workflow。
+     *
+     * @return 会进入挂起状态的 Workflow 定义
+     */
+    private ChainDefinition createConfirmDefinition() {
+        ChainDefinition definition = new ChainDefinition();
+        definition.setId("non-suspending-confirm-test");
+        StartNode start = new StartNode();
+        start.setId("start");
+        ConfirmNode confirm = new ConfirmNode();
+        confirm.setId("confirm");
+        EndNode end = new EndNode();
+        end.setId("end");
+        Edge first = new Edge();
+        first.setId("start-to-confirm");
+        first.setSource("start");
+        first.setTarget("confirm");
+        Edge second = new Edge();
+        second.setId("confirm-to-end");
+        second.setSource("confirm");
+        second.setTarget("end");
+        definition.addNode(start);
+        definition.addNode(confirm);
+        definition.addNode(end);
+        definition.addEdge(first);
+        definition.addEdge(second);
+        return definition;
+    }
+
    /**
     * 创建用于取消传播验证的工作流。
     *