diff --git a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java index a4f2955a..476d461d 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/builtin/DelegateAgentTool.java @@ -384,12 +384,16 @@ public class DelegateAgentTool { @vip.mate.tool.ConcurrencyUnsafe("internally fans out to its own thread pool; outer executor must not double-parallelize") @Tool(description = """ - Delegate multiple tasks to different Agents in parallel (max 3). \ + Delegate multiple tasks to different Agents in parallel (max 8). \ Each task runs concurrently in an independent child session. \ Use this when you have multiple independent sub-tasks that can run simultaneously. \ - Input is a JSON array: [{"agentName":"Agent名称","task":"任务描述"}, ...]""") + Input is a JSON array of objects with required "agentName" and "task", plus optional \ + "optional" (true = this task's failure must not abort the batch) and "timeout_seconds" \ + (widen the shared batch budget for a deliberately long task). When a required (non-optional) \ + task fails, remaining tasks are cancelled early instead of waiting out the full budget. \ + Example: [{"agentName":"X","task":"Y","optional":false,"timeout_seconds":120}, ...]""") public String delegateParallel( - @ToolParam(description = "JSON array of tasks: [{\"agentName\":\"X\",\"task\":\"Y\"}, ...]") + @ToolParam(description = "JSON array of tasks: [{\"agentName\":\"X\",\"task\":\"Y\",\"optional\":false,\"timeout_seconds\":120}, ...]") String tasksJson, // RFC-063r §2.5 改动点 5: hidden from LLM, used to inherit ChatOrigin into children. @Nullable ToolContext ctx) { @@ -433,9 +437,11 @@ public class DelegateAgentTool { // 2. Main thread: validate agents, create child conversations, register relays record PreparedChild(int index, AgentEntity agent, String task, String childConvId, - Runnable stopRelay, String subagentId) {} + Runnable stopRelay, String subagentId, boolean optional) {} List prepared = new ArrayList<>(); List errors = new ArrayList<>(); + // Highest per-task timeout override; widens the batch budget below. + int maxTaskTimeoutSeconds = 0; for (int i = 0; i < tasks.size(); i++) { Map t = tasks.get(i); @@ -453,6 +459,14 @@ public class DelegateAgentTool { continue; } + // Optional tasks never trigger fail-fast; per-task timeout_seconds + // (when present) widens the shared batch budget. + boolean optional = "true".equalsIgnoreCase(t.get("optional")); + Integer taskTimeout = parsePositiveIntOrNull(t.get("timeout_seconds")); + if (taskTimeout != null) { + maxTaskTimeoutSeconds = Math.max(maxTaskTimeoutSeconds, taskTimeout); + } + String childConvId = createChildConv(agent, parentConversationId); String subagentId = parentConversationId != null ? subagentRegistry.register(parentConversationId, childConvId, @@ -462,7 +476,7 @@ public class DelegateAgentTool { ? registerBatchedRelay(childConvId, rootConvFinal, agent.getName(), subagentId, parentSubagentId, childDepth) : null; - prepared.add(new PreparedChild(i, agent, task, childConvId, stopRelay, subagentId)); + prepared.add(new PreparedChild(i, agent, task, childConvId, stopRelay, subagentId, optional)); } if (prepared.isEmpty()) { @@ -484,6 +498,14 @@ public class DelegateAgentTool { "children", childrenInfo)); } + // Batch budget: the global default, widened by the largest per-task + // timeout_seconds override so a deliberately long child isn't cut off. + final int effectiveTimeoutSeconds = Math.max(parallelTimeoutSeconds, maxTaskTimeoutSeconds); + // Completes as soon as any REQUIRED child finishes unsuccessfully, so the + // wait below can collapse early (fail-fast) instead of burning the full + // budget while the parent already knows the batch can't succeed. + CompletableFuture requiredFailure = new CompletableFuture<>(); + // 4. Fan out — execute children in parallel long startTime = System.currentTimeMillis(); Map> futures = new LinkedHashMap<>(); @@ -528,19 +550,31 @@ public class DelegateAgentTool { }); } + // Required children arm the fail-fast signal on unsuccessful completion. + if (!p.optional()) { + future.thenAccept(r -> { + if (r != null && !r.success) { + requiredFailure.complete(null); + } + }); + } + futures.put(p.index, future); } - // 5. Wait for all children (with timeout) + // 5. Wait for all children, or bail out early when a required child fails. List results = new ArrayList<>(); try { - CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])) - .get(parallelTimeoutSeconds, TimeUnit.SECONDS); + CompletableFuture allDone = CompletableFuture.allOf( + futures.values().toArray(new CompletableFuture[0])); + CompletableFuture.anyOf(allDone, requiredFailure) + .get(effectiveTimeoutSeconds, TimeUnit.SECONDS); } catch (TimeoutException e) { - log.warn("Parallel delegation timed out ({}s), collecting completed results", parallelTimeoutSeconds); + log.warn("Parallel delegation timed out ({}s), collecting completed results", effectiveTimeoutSeconds); } catch (Exception e) { log.error("Parallel delegation error: {}", e.getMessage()); } + boolean failFast = requiredFailure.isDone(); // Collect results — completed futures get their value; unfinished ones are cancelled and recorded as timeout for (var entry : futures.entrySet()) { @@ -565,8 +599,11 @@ public class DelegateAgentTool { streamTracker.requestStop(p.childConvId); } f.cancel(true); - // Use ofTimeout so outcome="timeout" is explicit and distinct from "error". - results.add(ChildResult.ofTimeout(idx, agentName, parallelTimeoutSeconds)); + // Distinguish fail-fast cancellation from a genuine timeout so the + // parent doesn't misread a cancelled sibling as a slow agent. + results.add(failFast + ? ChildResult.ofCancelled(idx, agentName) + : ChildResult.ofTimeout(idx, agentName, effectiveTimeoutSeconds)); } } @@ -627,6 +664,7 @@ public class DelegateAgentTool { long successCount = results.stream().filter(r -> r.success && !r.isBlank()).count(); long blankCount = results.stream().filter(ChildResult::isBlank).count(); long timeoutCount = results.stream().filter(r -> "timeout".equals(r.outcome)).count(); + long cancelledCount = results.stream().filter(r -> "cancelled".equals(r.outcome)).count(); long errorCount = results.stream().filter(r -> "error".equals(r.outcome)).count(); StringBuilder sb = new StringBuilder(); @@ -639,6 +677,7 @@ public class DelegateAgentTool { .append(" success=").append(successCount) .append(" blank_success=").append(blankCount) .append(" timeout=").append(timeoutCount) + .append(" cancelled=").append(cancelledCount) .append(" error=").append(errorCount) .append(" durationMs=").append(totalDurationMs) .append("\n\n"); @@ -673,7 +712,7 @@ public class DelegateAgentTool { .append("，trim 后 0 字符）。请勿将此误报为超时或失败——子 Agent 已正常完成，只是本次无输出。\n"); } case "timeout" -> - sb.append("❌ 超时（").append(parallelTimeoutSeconds).append("s 内未返回）\n"); + sb.append("❌ 超时（").append(effectiveTimeoutSeconds).append("s 内未返回）\n"); default -> sb.append("❌ 失败：").append(r.error).append("\n"); } @@ -1084,6 +1123,16 @@ public class DelegateAgentTool { "timeout", 0, 0); } + /** + * Factory for a child cancelled by fail-fast — a required sibling failed, + * so this still-running child was stopped before finishing. Distinct from + * a timeout (it was not slow; the batch was abandoned). + */ + static ChildResult ofCancelled(int idx, String name) { + return new ChildResult(idx, name, false, null, + "已取消（必需子任务失败，触发提前收束）", 0, "cancelled", 0, 0); + } + // Legacy shims — kept for callers that pre-date the factory methods static ChildResult success(int idx, String name, String result, long ms) { // result may already be truncated at call site — lengths will be approximate @@ -1336,6 +1385,17 @@ public class DelegateAgentTool { return DelegationContext.parentConversationId(); } + /** Parse a positive integer (seconds) or return null for blank / invalid / non-positive input. */ + private static Integer parsePositiveIntOrNull(String raw) { + if (raw == null || raw.isBlank()) return null; + try { + int v = Integer.parseInt(raw.trim()); + return v > 0 ? v : null; + } catch (NumberFormatException e) { + return null; + } + } + private String availableAgentsHint() { List agents = agentMapper.selectList(new LambdaQueryWrapper() .eq(AgentEntity::getEnabled, true).select(AgentEntity::getName)); diff --git a/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAgentToolTest.java b/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAgentToolTest.java index a8e6d1ec..33be4673 100644 --- a/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAgentToolTest.java +++ b/mateclaw-server/src/test/java/vip/mate/tool/builtin/DelegateAgentToolTest.java @@ -263,4 +263,109 @@ class DelegateAgentToolTest { assertTrue(result.contains("超时") || result.contains("✗"), "Should contain timeout indicator for SlowAgent: " + result); } + + // ===== delegateParallel: fail-fast on required failure (RFC 05 Q3) ===== + + @Test + @DisplayName("delegateParallel fails fast: required failure cancels a slow sibling") + void delegateParallelRequiredFailureCancelsSibling() { + AgentEntity failAgent = new AgentEntity(); + failAgent.setId(20L); + failAgent.setName("FailAgent"); + failAgent.setEnabled(true); + failAgent.setWorkspaceId(1L); + + AgentEntity slowAgent = new AgentEntity(); + slowAgent.setId(21L); + slowAgent.setName("SlowAgent"); + slowAgent.setEnabled(true); + slowAgent.setWorkspaceId(1L); + + when(agentMapper.selectOne(any(LambdaQueryWrapper.class))) + .thenReturn(failAgent) + .thenReturn(slowAgent); + when(streamTracker.isRunning(any())).thenReturn(false); + // Required FailAgent errors immediately → arms fail-fast. + when(agentService.chat(eq(20L), anyString(), anyString(), any())) + .thenThrow(new RuntimeException("boom")); + // SlowAgent would block well past the 3 s test budget; fail-fast cancels it. + when(agentService.chat(eq(21L), anyString(), anyString(), any())).thenAnswer(inv -> { + Thread.sleep(10_000); + return "unreachable"; + }); + + ToolExecutionContext.set("parent-ff", "admin"); + String json = "[{\"agentName\":\"FailAgent\",\"task\":\"a\"},{\"agentName\":\"SlowAgent\",\"task\":\"b\"}]"; + + long start = System.currentTimeMillis(); + String result = delegateAgentTool.delegateParallel(json, null); + long elapsed = System.currentTimeMillis() - start; + + assertTrue(elapsed < 2500, "fail-fast should return well before the budget, took " + elapsed + "ms"); + assertTrue(result.contains("cancelled=1"), "slow sibling should be cancelled: " + result); + assertTrue(result.contains("已取消"), "should label the cancelled sibling: " + result); + } + + @Test + @DisplayName("delegateParallel: optional task failure does not abort the batch") + void delegateParallelOptionalFailureDoesNotAbort() { + AgentEntity optAgent = new AgentEntity(); + optAgent.setId(30L); + optAgent.setName("OptAgent"); + optAgent.setEnabled(true); + optAgent.setWorkspaceId(1L); + + AgentEntity okAgent = new AgentEntity(); + okAgent.setId(31L); + okAgent.setName("OkAgent"); + okAgent.setEnabled(true); + okAgent.setWorkspaceId(1L); + + when(agentMapper.selectOne(any(LambdaQueryWrapper.class))) + .thenReturn(optAgent) + .thenReturn(okAgent); + when(streamTracker.isRunning(any())).thenReturn(false); + when(agentService.chat(eq(30L), anyString(), anyString(), any())) + .thenThrow(new RuntimeException("opt boom")); + when(agentService.chat(eq(31L), anyString(), anyString(), any())) + .thenReturn("ok result done"); + + ToolExecutionContext.set("parent-opt", "admin"); + String json = "[{\"agentName\":\"OptAgent\",\"task\":\"a\",\"optional\":true}," + + "{\"agentName\":\"OkAgent\",\"task\":\"b\"}]"; + String result = delegateAgentTool.delegateParallel(json, null); + + // An optional failure must not cancel anything; the other task completes normally. + assertTrue(result.contains("cancelled=0"), "no cancellation expected: " + result); + assertTrue(result.contains("OkAgent"), "ok task should be reported: " + result); + } + + // ===== delegateParallel: per-task timeout override (RFC 05 Q2) ===== + + @Test + @DisplayName("delegateParallel: per-task timeout_seconds widens the batch budget") + void delegateParallelTimeoutOverrideWidensBudget() { + AgentEntity longAgent = new AgentEntity(); + longAgent.setId(40L); + longAgent.setName("LongAgent"); + longAgent.setEnabled(true); + longAgent.setWorkspaceId(1L); + + when(agentMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(longAgent); + when(streamTracker.isRunning(any())).thenReturn(false); + // Sleeps 4 s — beyond the 3 s test budget, but within the 6 s override. + when(agentService.chat(eq(40L), anyString(), anyString(), any())).thenAnswer(inv -> { + Thread.sleep(4_000); + return "long task finished ok"; + }); + + ToolExecutionContext.set("parent-to", "admin"); + String json = "[{\"agentName\":\"LongAgent\",\"task\":\"a\",\"timeout_seconds\":6}]"; + String result = delegateAgentTool.delegateParallel(json, null); + + // With the override the child finishes instead of timing out at 3 s. + assertTrue(result.contains("long task finished ok") || result.contains("success=1"), + "long task should complete within the widened budget: " + result); + assertFalse(result.contains("timeout=1"), "should not time out with the override: " + result); + } }