mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
feat(delegation): 子 Agent 成本透出 + 单任务/计划步骤委派结构化
- 子执行改走 chatWithUsage，捕获并透出每个子 Agent 的 prompt/completion token （单任务回复、并行机读头+逐行、delegation_end/child_complete/单路 broadcastEnd 事件） - 新增 delegateByAgentIdStructured 返回结构化 ChildResult；计划步骤委派改按 success()/isBlank() 判定成败，替掉脆弱的错误前缀匹配 - 前端委派段与嵌套节点显示紧凑成本后缀/徽标 - 测试：子执行 stub 迁移到 chatWithUsage + 新增 token 回归用例
This commit is contained in:
matevip
parent
db11433883
commit
d390935763
@ -30,6 +30,7 @@ import vip.mate.planning.service.PlanningService;
|
||||
import vip.mate.agent.context.ChatOrigin;
|
||||
import vip.mate.skill.runtime.SkillCatalogRenderer;
|
||||
import vip.mate.tool.builtin.DelegateAgentTool;
|
||||
import vip.mate.tool.builtin.DelegateAgentTool.ChildResult;
|
||||
import vip.mate.tool.builtin.DelegationContext;
|
||||
import vip.mate.tool.builtin.ToolExecutionContext;
|
||||
|
||||
@ -678,7 +679,7 @@ public class StepExecutionNode implements NodeAction {
|
||||
// Seed the delegation context with the plan's REAL conversation id (from
|
||||
// graph state) so the delegated child conversation is parented to it and
|
||||
// stays hidden from the user's conversation list. The ChatOrigin in the
|
||||
// plan-execute path carries no conversationId, so delegateByAgentId can't
|
||||
// plan-execute path carries no conversationId, so the delegation can't
|
||||
// derive the parent on its own — we provide it here.
|
||||
boolean seeded = false;
|
||||
if (conversationId != null && !conversationId.isBlank()
|
||||
@ -687,20 +688,30 @@ public class StepExecutionNode implements NodeAction {
|
||||
DelegationContext.enter(conversationId, Set.of(), conversationId, null, 0);
|
||||
seeded = true;
|
||||
}
|
||||
String result;
|
||||
ChildResult childResult = null;
|
||||
String delegateError = null;
|
||||
try {
|
||||
result = delegateAgentTool.delegateByAgentId(assignedAgentId, step, chatOrigin);
|
||||
childResult = delegateAgentTool.delegateByAgentIdStructured(assignedAgentId, step, chatOrigin);
|
||||
} catch (Exception e) {
|
||||
log.error("[StepExecution] Delegated step {} threw: {}", stepIndex, e.getMessage(), e);
|
||||
result = "[错误] 委派执行异常：" + e.getMessage();
|
||||
delegateError = e.getMessage();
|
||||
} finally {
|
||||
if (seeded) {
|
||||
DelegationContext.exit();
|
||||
}
|
||||
}
|
||||
|
||||
String finalResult = result != null ? result : "";
|
||||
boolean failed = finalResult.isEmpty() || finalResult.startsWith("[错误]");
|
||||
// Branch on the structured outcome instead of pattern-matching an error
|
||||
// prefix out of the reply text: a successful child with non-empty content
|
||||
// is the only "ok" case; blank / error / missing all count as failure.
|
||||
boolean ok = childResult != null && childResult.success() && !childResult.isBlank();
|
||||
String finalResult = ok
|
||||
? (childResult.result() != null ? childResult.result() : "")
|
||||
: "[错误] 委派执行失败：" + (delegateError != null ? delegateError
|
||||
: childResult != null && childResult.error() != null ? childResult.error()
|
||||
: childResult != null && childResult.isBlank() ? "子 Agent 返回内容为空"
|
||||
: "未知错误");
|
||||
boolean failed = !ok;
|
||||
if (failed) {
|
||||
planningService.updateSubPlanFailure(planId, stepIndex, finalResult);
|
||||
} else {
|
||||
|
||||
@ -13,6 +13,7 @@ import org.springframework.lang.Nullable;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.AgentService.ChatResult;
|
||||
import vip.mate.agent.context.ChatOrigin;
|
||||
import vip.mate.agent.delegation.SubagentRegistry;
|
||||
import vip.mate.agent.model.AgentEntity;
|
||||
@ -242,6 +243,38 @@ public class DelegateAgentTool {
|
||||
return "[错误] 未找到名为「" + agentName + "」的已启用 Agent。" + availableAgentsHint();
|
||||
}
|
||||
|
||||
// Execute via the shared single-task core (registry, relay, broadcast,
|
||||
// child run). Returns the structured ChildResult plus the child's
|
||||
// session handle so the parent can follow up on this exact sub-agent.
|
||||
SingleDelegation sd = executeSingleDelegation(target, task, inheritParentContext, ctx);
|
||||
ChildResult result = sd.result();
|
||||
|
||||
String response = result.toToolResponse(target.getName());
|
||||
// Surface the child's session handle so the parent can follow up on this
|
||||
// exact sub-agent (its conversation persists past this call) via
|
||||
// send_to_subagent, instead of re-spawning a fresh, context-less child.
|
||||
if (result.success() && sd.childConversationId() != null) {
|
||||
response += "\n\n[session_id: " + sd.childConversationId()
|
||||
+ " — to follow up with this sub-agent, call send_to_subagent(session_id, message)]";
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
/** Carrier for a single-task delegation: the structured child result plus
|
||||
* the child conversation handle (null when spawning was short-circuited). */
|
||||
private record SingleDelegation(ChildResult result, String childConversationId) {}
|
||||
|
||||
/**
|
||||
* Shared execution core for single-task delegation, used by both the
|
||||
* LLM-facing {@link #delegateToAgent} tool and the id-based
|
||||
* {@link #delegateByAgentIdStructured} (per-step plan delegation). Handles
|
||||
* spawn-pause, child conversation creation, optional parent-context
|
||||
* inheritance, sub-agent registry, event relay/broadcast, and the child
|
||||
* run — returning the structured {@link ChildResult} so callers decide how
|
||||
* to format it (tool string vs. plan step bookkeeping).
|
||||
*/
|
||||
private SingleDelegation executeSingleDelegation(AgentEntity target, String task,
|
||||
Boolean inheritParentContext, ToolContext ctx) {
|
||||
String parentConversationId = resolveParentConversationId();
|
||||
// Root (human-facing) conversation at the top of the delegation tree.
|
||||
// At depth 0 the immediate parent IS the root; deeper layers carry it
|
||||
@ -249,14 +282,15 @@ public class DelegateAgentTool {
|
||||
String rootConversationId = DelegationContext.rootConversationId();
|
||||
if (rootConversationId == null) rootConversationId = parentConversationId;
|
||||
String parentSubagentId = DelegationContext.currentSubagentId();
|
||||
int childDepth = depth + 1;
|
||||
int childDepth = DelegationContext.currentDepth() + 1;
|
||||
|
||||
// Spawn-pause: short-circuit before creating child state when either the
|
||||
// immediate parent or the root tree is paused, so no conversation rows /
|
||||
// relays / registry entries leak.
|
||||
if ((parentConversationId != null && subagentRegistry.isSpawnPaused(parentConversationId))
|
||||
|| (rootConversationId != null && subagentRegistry.isSpawnPaused(rootConversationId))) {
|
||||
return "[错误] Spawning paused for this conversation; resume via /api/v1/subagents/spawn-pause";
|
||||
return new SingleDelegation(ChildResult.ofError(0, target.getName(),
|
||||
"Spawning paused for this conversation; resume via /api/v1/subagents/spawn-pause"), null);
|
||||
}
|
||||
|
||||
String childConversationId = createChildConv(target, parentConversationId);
|
||||
@ -275,12 +309,12 @@ public class DelegateAgentTool {
|
||||
}
|
||||
|
||||
log.info("Agent delegation: depth={}, target={}({}), childConv={}, parentConv={}",
|
||||
depth + 1, target.getName(), target.getId(), childConversationId, parentConversationId);
|
||||
childDepth, target.getName(), target.getId(), childConversationId, parentConversationId);
|
||||
|
||||
// Register the live sub-agent first so its stable id rides on every
|
||||
// event. Disposable is null in the synchronous single-task path because
|
||||
// the executor blocks on AgentService#chat directly — there is no Flux
|
||||
// subscription to dispose. Interrupts here are best-effort (status flip).
|
||||
// the executor blocks on AgentService#chatWithUsage directly — there is
|
||||
// no Flux subscription to dispose. Interrupts here are best-effort (status flip).
|
||||
String subagentId = parentConversationId != null
|
||||
? subagentRegistry.register(parentConversationId, childConversationId,
|
||||
target.getId(), task, null, parentSubagentId, childDepth, rootConversationId)
|
||||
@ -326,16 +360,7 @@ public class DelegateAgentTool {
|
||||
broadcastEnd(rootConversationId, childConversationId, target.getName(), result,
|
||||
subagentId, parentSubagentId, childDepth);
|
||||
}
|
||||
|
||||
String response = result.toToolResponse(target.getName());
|
||||
// Surface the child's session handle so the parent can follow up on this
|
||||
// exact sub-agent (its conversation persists past this call) via
|
||||
// send_to_subagent, instead of re-spawning a fresh, context-less child.
|
||||
if (result.success() && childConversationId != null) {
|
||||
response += "\n\n[session_id: " + childConversationId
|
||||
+ " — to follow up with this sub-agent, call send_to_subagent(session_id, message)]";
|
||||
}
|
||||
return response;
|
||||
return new SingleDelegation(result, childConversationId);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -347,12 +372,30 @@ public class DelegateAgentTool {
|
||||
* child's reply text, or an error string when the agent is missing/disabled.
|
||||
*/
|
||||
public String delegateByAgentId(Long agentId, String task, ChatOrigin parentOrigin) {
|
||||
ChildResult result = delegateByAgentIdStructured(agentId, task, parentOrigin);
|
||||
return result.toToolResponse(result.agentName());
|
||||
}
|
||||
|
||||
/**
|
||||
* Structured variant of {@link #delegateByAgentId}: runs the same isolated
|
||||
* child execution but returns the full {@link ChildResult} instead of a
|
||||
* formatted string. Used by per-step plan delegation so the plan graph can
|
||||
* branch on {@code success()} / {@code isBlank()} / {@code outcome()} and
|
||||
* read token usage, rather than pattern-matching an error prefix out of a
|
||||
* string. Never throws — agent-resolution and depth-guard failures come
|
||||
* back as {@code outcome="error"} results.
|
||||
*/
|
||||
public ChildResult delegateByAgentIdStructured(Long agentId, String task, ChatOrigin parentOrigin) {
|
||||
if (agentId == null) {
|
||||
return "[错误] 未指定委派 Agent。";
|
||||
return ChildResult.ofError(0, "?", "未指定委派 Agent。");
|
||||
}
|
||||
AgentEntity target = agentMapper.selectById(agentId);
|
||||
if (target == null || !Boolean.TRUE.equals(target.getEnabled())) {
|
||||
return "[错误] 未找到 id=" + agentId + " 的已启用 Agent。";
|
||||
return ChildResult.ofError(0, "id=" + agentId, "未找到 id=" + agentId + " 的已启用 Agent。");
|
||||
}
|
||||
if (DelegationContext.currentDepth() >= MAX_DELEGATION_DEPTH) {
|
||||
return ChildResult.ofError(0, target.getName(),
|
||||
"委派层级已达上限（" + MAX_DELEGATION_DEPTH + " 层）");
|
||||
}
|
||||
ChatOrigin origin = parentOrigin != null ? parentOrigin : ChatOrigin.EMPTY;
|
||||
ToolContext ctx = origin.toToolContext();
|
||||
@ -372,7 +415,7 @@ public class DelegateAgentTool {
|
||||
DelegationContext.enter(parentConvId, Set.of(), parentConvId, null, 0);
|
||||
}
|
||||
try {
|
||||
return delegateToAgent(target.getName(), task, false, ctx);
|
||||
return executeSingleDelegation(target, task, false, ctx).result();
|
||||
} finally {
|
||||
if (seedContext) {
|
||||
DelegationContext.exit();
|
||||
@ -543,6 +586,8 @@ public class DelegateAgentTool {
|
||||
payload.put("trimmedLength", r.trimmedLength);
|
||||
payload.put("blank", r.isBlank());
|
||||
payload.put("durationMs", r.durationMs);
|
||||
payload.put("promptTokens", r.promptTokens);
|
||||
payload.put("completionTokens", r.completionTokens);
|
||||
payload.put("resultPreview", r.success
|
||||
? truncate(r.result, 400)
|
||||
: (r.error != null ? r.error : "error"));
|
||||
@ -637,6 +682,8 @@ public class DelegateAgentTool {
|
||||
m.put("trimmedLength", r.trimmedLength);
|
||||
m.put("blank", r.isBlank());
|
||||
m.put("durationMs", r.durationMs);
|
||||
m.put("promptTokens", r.promptTokens);
|
||||
m.put("completionTokens", r.completionTokens);
|
||||
// childConversationId + subagentId for stable frontend tree lookup
|
||||
prepared.stream()
|
||||
.filter(p -> p.index == r.taskIndex)
|
||||
@ -666,6 +713,8 @@ public class DelegateAgentTool {
|
||||
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();
|
||||
long tokensInTotal = results.stream().mapToLong(r -> r.promptTokens).sum();
|
||||
long tokensOutTotal = results.stream().mapToLong(r -> r.completionTokens).sum();
|
||||
|
||||
StringBuilder sb = new StringBuilder();
|
||||
|
||||
@ -680,6 +729,8 @@ public class DelegateAgentTool {
|
||||
.append(" cancelled=").append(cancelledCount)
|
||||
.append(" error=").append(errorCount)
|
||||
.append(" durationMs=").append(totalDurationMs)
|
||||
.append(" tokensIn=").append(tokensInTotal)
|
||||
.append(" tokensOut=").append(tokensOutTotal)
|
||||
.append("\n\n");
|
||||
|
||||
// Important: this result is from the current execution. Any timeout entries in the
|
||||
@ -700,6 +751,8 @@ public class DelegateAgentTool {
|
||||
.append(" | contentLength=").append(r.trimmedLength).append("chars")
|
||||
.append(" | rawLength=").append(r.rawLength).append("chars")
|
||||
.append(" | duration=").append(r.durationMs / 1000).append("s")
|
||||
.append(" | tokensIn=").append(r.promptTokens)
|
||||
.append(" | tokensOut=").append(r.completionTokens)
|
||||
.append("\n\n");
|
||||
|
||||
switch (r.outcome) {
|
||||
@ -1048,11 +1101,17 @@ public class DelegateAgentTool {
|
||||
ChatOrigin childOrigin = (parentOrigin != null ? parentOrigin : ChatOrigin.EMPTY)
|
||||
.withAgent(target.getId())
|
||||
.withConversationId(childConversationId);
|
||||
String rawResult = agentService.chat(target.getId(), task, childConversationId, childOrigin);
|
||||
// chatWithUsage runs the same StateGraph as chat() (so the child
|
||||
// message is persisted identically) but also surfaces the child's
|
||||
// token usage from the graph's _usage_final event, so the parent can
|
||||
// see what each sub-agent cost.
|
||||
ChatResult chatResult = agentService.chatWithUsage(
|
||||
target.getId(), task, childConversationId, childOrigin);
|
||||
long durationMs = System.currentTimeMillis() - startTime;
|
||||
String rawResult = chatResult.content();
|
||||
// Measure lengths before truncation so ChildResult carries accurate metadata.
|
||||
return ChildResult.ofSuccess(taskIndex, target.getName(), rawResult, durationMs,
|
||||
MAX_RESULT_LENGTH);
|
||||
MAX_RESULT_LENGTH, chatResult.promptTokens(), chatResult.completionTokens());
|
||||
} catch (Exception e) {
|
||||
log.error("Child agent failed: taskIndex={}, agent={}, error={}",
|
||||
taskIndex, target.getName(), e.getMessage());
|
||||
@ -1079,21 +1138,25 @@ public class DelegateAgentTool {
|
||||
* <p>{@code rawLength} and {@code trimmedLength} are measured before truncation and reflect the
|
||||
* true content length.
|
||||
*/
|
||||
private record ChildResult(
|
||||
public record ChildResult(
|
||||
int taskIndex, String agentName, boolean success,
|
||||
String result, String error, long durationMs,
|
||||
/** "success" | "blank_success" | "timeout" | "error" */
|
||||
/** "success" | "blank_success" | "timeout" | "cancelled" | "error" */
|
||||
String outcome,
|
||||
int rawLength, int trimmedLength) {
|
||||
int rawLength, int trimmedLength,
|
||||
/** Child token usage captured from the graph's _usage_final event;
|
||||
* 0 for non-success outcomes (timeout / error / cancelled). */
|
||||
int promptTokens, int completionTokens) {
|
||||
|
||||
/** Whether the child returned no usable content (blank_success). */
|
||||
boolean isBlank() { return "blank_success".equals(outcome); }
|
||||
public boolean isBlank() { return "blank_success".equals(outcome); }
|
||||
|
||||
/**
|
||||
* Factory for a successful child execution.
|
||||
* Measures lengths from the raw result before applying the truncation limit.
|
||||
*/
|
||||
static ChildResult ofSuccess(int idx, String name, String rawResult, long ms, int maxLen) {
|
||||
static ChildResult ofSuccess(int idx, String name, String rawResult, long ms, int maxLen,
|
||||
int promptTokens, int completionTokens) {
|
||||
String safe = rawResult != null ? rawResult : "";
|
||||
String trimmed = safe.trim();
|
||||
boolean blank = trimmed.isEmpty();
|
||||
@ -1102,7 +1165,8 @@ public class DelegateAgentTool {
|
||||
truncate(safe, maxLen),
|
||||
null, ms,
|
||||
blank ? "blank_success" : "success",
|
||||
safe.length(), trimmed.length());
|
||||
safe.length(), trimmed.length(),
|
||||
Math.max(0, promptTokens), Math.max(0, completionTokens));
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1113,14 +1177,14 @@ public class DelegateAgentTool {
|
||||
String msg = err != null ? err : "Unknown error";
|
||||
boolean isTimeout = msg.contains("超时") || msg.toLowerCase().contains("timeout");
|
||||
return new ChildResult(idx, name, false, null, msg, 0,
|
||||
isTimeout ? "timeout" : "error", 0, 0);
|
||||
isTimeout ? "timeout" : "error", 0, 0, 0, 0);
|
||||
}
|
||||
|
||||
/** Factory for an explicit timeout (parallel window exceeded). */
|
||||
static ChildResult ofTimeout(int idx, String name, int timeoutSec) {
|
||||
String msg = "超时 (" + timeoutSec + "s)";
|
||||
return new ChildResult(idx, name, false, null, msg, (long) timeoutSec * 1000L,
|
||||
"timeout", 0, 0);
|
||||
"timeout", 0, 0, 0, 0);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1130,7 +1194,7 @@ public class DelegateAgentTool {
|
||||
*/
|
||||
static ChildResult ofCancelled(int idx, String name) {
|
||||
return new ChildResult(idx, name, false, null,
|
||||
"已取消（必需子任务失败，触发提前收束）", 0, "cancelled", 0, 0);
|
||||
"已取消（必需子任务失败，触发提前收束）", 0, "cancelled", 0, 0, 0, 0);
|
||||
}
|
||||
|
||||
// Legacy shims — kept for callers that pre-date the factory methods
|
||||
@ -1140,15 +1204,19 @@ public class DelegateAgentTool {
|
||||
String trimmed = safe.trim();
|
||||
boolean blank = trimmed.isEmpty();
|
||||
return new ChildResult(idx, name, true, safe, null, ms,
|
||||
blank ? "blank_success" : "success", safe.length(), trimmed.length());
|
||||
blank ? "blank_success" : "success", safe.length(), trimmed.length(), 0, 0);
|
||||
}
|
||||
static ChildResult error(int idx, String name, String err) {
|
||||
return ofError(idx, name, err);
|
||||
}
|
||||
|
||||
String toToolResponse(String agentName) {
|
||||
if (success) return "[Agent「" + agentName + "」的回复]\n\n" + (result != null ? result : "");
|
||||
return "[错误] Agent「" + agentName + "」执行失败: " + error;
|
||||
if (!success) return "[错误] Agent「" + agentName + "」执行失败: " + error;
|
||||
String body = "[Agent「" + agentName + "」的回复]\n\n" + (result != null ? result : "");
|
||||
if (promptTokens > 0 || completionTokens > 0) {
|
||||
body += "\n\n[usage: tokensIn=" + promptTokens + " tokensOut=" + completionTokens + "]";
|
||||
}
|
||||
return body;
|
||||
}
|
||||
|
||||
private static String truncate(String text, int maxLength) {
|
||||
@ -1372,6 +1440,8 @@ public class DelegateAgentTool {
|
||||
Map<String, Object> ev = delegationPayload(subagentId, parentSubagentId, depth, childConvId, agentName);
|
||||
ev.put("success", result.success);
|
||||
ev.put("durationMs", result.durationMs);
|
||||
ev.put("promptTokens", result.promptTokens);
|
||||
ev.put("completionTokens", result.completionTokens);
|
||||
ev.put("resultPreview",
|
||||
result.success ? truncate(result.result, 200) : (result.error != null ? result.error : ""));
|
||||
streamTracker.broadcastObject(rootConvId, "delegation_end", ev);
|
||||
|
||||
@ -15,6 +15,7 @@ import org.mockito.Mock;
|
||||
import org.mockito.Spy;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.AgentService.ChatResult;
|
||||
import vip.mate.agent.delegation.SubagentRegistry;
|
||||
import vip.mate.agent.model.AgentEntity;
|
||||
import vip.mate.agent.repository.AgentMapper;
|
||||
@ -180,9 +181,9 @@ class DelegateAgentToolTest {
|
||||
// the test thread returns immediately. The orphan keeps sleeping on a
|
||||
// virtual thread until JVM teardown — that's the same behavior as
|
||||
// production (cancel is best-effort).
|
||||
when(agentService.chat(anyLong(), anyString(), anyString(), any())).thenAnswer(invocation -> {
|
||||
when(agentService.chatWithUsage(anyLong(), anyString(), anyString(), any())).thenAnswer(invocation -> {
|
||||
Thread.sleep(10_000);
|
||||
return "should not reach here";
|
||||
return ChatResult.contentOnly("should not reach here");
|
||||
});
|
||||
|
||||
// Set a conversationId so resolveParentConversationId works
|
||||
@ -239,13 +240,13 @@ class DelegateAgentToolTest {
|
||||
when(streamTracker.isRunning(any())).thenReturn(false);
|
||||
|
||||
// FastAgent completes immediately
|
||||
when(agentService.chat(eq(10L), anyString(), anyString(), any()))
|
||||
.thenReturn("Fast result completed successfully");
|
||||
when(agentService.chatWithUsage(eq(10L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("Fast result completed successfully"));
|
||||
|
||||
// SlowAgent blocks longer than the (test-overridden) 3 s budget.
|
||||
when(agentService.chat(eq(11L), anyString(), anyString(), any())).thenAnswer(invocation -> {
|
||||
when(agentService.chatWithUsage(eq(11L), anyString(), anyString(), any())).thenAnswer(invocation -> {
|
||||
Thread.sleep(10_000);
|
||||
return "should not reach here";
|
||||
return ChatResult.contentOnly("should not reach here");
|
||||
});
|
||||
|
||||
ToolExecutionContext.set("parent-mixed", "admin");
|
||||
@ -286,12 +287,12 @@ class DelegateAgentToolTest {
|
||||
.thenReturn(slowAgent);
|
||||
when(streamTracker.isRunning(any())).thenReturn(false);
|
||||
// Required FailAgent errors immediately → arms fail-fast.
|
||||
when(agentService.chat(eq(20L), anyString(), anyString(), any()))
|
||||
when(agentService.chatWithUsage(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 -> {
|
||||
when(agentService.chatWithUsage(eq(21L), anyString(), anyString(), any())).thenAnswer(inv -> {
|
||||
Thread.sleep(10_000);
|
||||
return "unreachable";
|
||||
return ChatResult.contentOnly("unreachable");
|
||||
});
|
||||
|
||||
ToolExecutionContext.set("parent-ff", "admin");
|
||||
@ -325,10 +326,10 @@ class DelegateAgentToolTest {
|
||||
.thenReturn(optAgent)
|
||||
.thenReturn(okAgent);
|
||||
when(streamTracker.isRunning(any())).thenReturn(false);
|
||||
when(agentService.chat(eq(30L), anyString(), anyString(), any()))
|
||||
when(agentService.chatWithUsage(eq(30L), anyString(), anyString(), any()))
|
||||
.thenThrow(new RuntimeException("opt boom"));
|
||||
when(agentService.chat(eq(31L), anyString(), anyString(), any()))
|
||||
.thenReturn("ok result done");
|
||||
when(agentService.chatWithUsage(eq(31L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("ok result done"));
|
||||
|
||||
ToolExecutionContext.set("parent-opt", "admin");
|
||||
String json = "[{\"agentName\":\"OptAgent\",\"task\":\"a\",\"optional\":true},"
|
||||
@ -354,9 +355,9 @@ class DelegateAgentToolTest {
|
||||
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 -> {
|
||||
when(agentService.chatWithUsage(eq(40L), anyString(), anyString(), any())).thenAnswer(inv -> {
|
||||
Thread.sleep(4_000);
|
||||
return "long task finished ok";
|
||||
return ChatResult.contentOnly("long task finished ok");
|
||||
});
|
||||
|
||||
ToolExecutionContext.set("parent-to", "admin");
|
||||
@ -368,4 +369,60 @@ class DelegateAgentToolTest {
|
||||
"long task should complete within the widened budget: " + result);
|
||||
assertFalse(result.contains("timeout=1"), "should not time out with the override: " + result);
|
||||
}
|
||||
|
||||
// ===== token usage surfacing =====
|
||||
|
||||
@Test
|
||||
@DisplayName("delegateToAgent surfaces the child's token usage in the reply")
|
||||
void delegateToAgentSurfacesTokenUsage() {
|
||||
AgentEntity agent = new AgentEntity();
|
||||
agent.setId(50L);
|
||||
agent.setName("Worker");
|
||||
agent.setEnabled(true);
|
||||
agent.setWorkspaceId(1L);
|
||||
when(agentMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(agent);
|
||||
when(streamTracker.isRunning(any())).thenReturn(false);
|
||||
when(agentService.chatWithUsage(eq(50L), anyString(), anyString(), any()))
|
||||
.thenReturn(new ChatResult("done with work", 120, 45, null, null));
|
||||
|
||||
ToolExecutionContext.set("parent-usage", "admin");
|
||||
String result = delegateAgentTool.delegateToAgent("Worker", "do the thing", null, null);
|
||||
|
||||
assertTrue(result.contains("tokensIn=120"), "reply should surface prompt tokens: " + result);
|
||||
assertTrue(result.contains("tokensOut=45"), "reply should surface completion tokens: " + result);
|
||||
}
|
||||
|
||||
@Test
|
||||
@DisplayName("delegateParallel aggregates child token usage in the header and per-row lines")
|
||||
void delegateParallelAggregatesTokenUsage() {
|
||||
AgentEntity a = new AgentEntity();
|
||||
a.setId(60L);
|
||||
a.setName("AgentA");
|
||||
a.setEnabled(true);
|
||||
a.setWorkspaceId(1L);
|
||||
AgentEntity b = new AgentEntity();
|
||||
b.setId(61L);
|
||||
b.setName("AgentB");
|
||||
b.setEnabled(true);
|
||||
b.setWorkspaceId(1L);
|
||||
when(agentMapper.selectOne(any(LambdaQueryWrapper.class)))
|
||||
.thenReturn(a)
|
||||
.thenReturn(b);
|
||||
when(streamTracker.isRunning(any())).thenReturn(false);
|
||||
when(agentService.chatWithUsage(eq(60L), anyString(), anyString(), any()))
|
||||
.thenReturn(new ChatResult("result A", 100, 30, null, null));
|
||||
when(agentService.chatWithUsage(eq(61L), anyString(), anyString(), any()))
|
||||
.thenReturn(new ChatResult("result B", 80, 20, null, null));
|
||||
|
||||
ToolExecutionContext.set("parent-usage-parallel", "admin");
|
||||
String json = "[{\"agentName\":\"AgentA\",\"task\":\"a\"},{\"agentName\":\"AgentB\",\"task\":\"b\"}]";
|
||||
String result = delegateAgentTool.delegateParallel(json, null);
|
||||
|
||||
// Machine header carries the batch totals (180 in, 50 out).
|
||||
assertTrue(result.contains("tokensIn=180"), "header should aggregate prompt tokens: " + result);
|
||||
assertTrue(result.contains("tokensOut=50"), "header should aggregate completion tokens: " + result);
|
||||
// Per-row lines carry each child's own usage.
|
||||
assertTrue(result.contains("tokensIn=100"), "row A should carry its prompt tokens: " + result);
|
||||
assertTrue(result.contains("tokensIn=80"), "row B should carry its prompt tokens: " + result);
|
||||
}
|
||||
}
|
||||
|
||||
@ -16,6 +16,7 @@ import org.mockito.Mock;
|
||||
import org.mockito.junit.jupiter.MockitoExtension;
|
||||
import org.mockito.Spy;
|
||||
import vip.mate.agent.AgentService;
|
||||
import vip.mate.agent.AgentService.ChatResult;
|
||||
import vip.mate.agent.delegation.SubagentRegistry;
|
||||
import vip.mate.agent.model.AgentEntity;
|
||||
import vip.mate.agent.repository.AgentMapper;
|
||||
@ -108,14 +109,14 @@ class DelegateEventSequenceTest {
|
||||
});
|
||||
|
||||
// During chat(), simulate the child broadcasting a tool_call_started event
|
||||
when(agentService.chat(eq(100L), eq("summarize the report"), anyString(), any()))
|
||||
when(agentService.chatWithUsage(eq(100L), eq("summarize the report"), anyString(), any()))
|
||||
.thenAnswer(invocation -> {
|
||||
// The relay listener should have been registered by now — fire it
|
||||
BiConsumer<String, String> relay = relayRef.get();
|
||||
assertNotNull(relay, "Relay should be registered before child chat starts");
|
||||
relay.accept("tool_call_started", "{\"name\":\"searchWeb\"}");
|
||||
relay.accept("tool_call_completed", "{\"name\":\"searchWeb\",\"success\":true}");
|
||||
return "The report shows growth of 15% YoY.";
|
||||
return ChatResult.contentOnly("The report shows growth of 15% YoY.");
|
||||
});
|
||||
|
||||
// Act
|
||||
@ -170,8 +171,10 @@ class DelegateEventSequenceTest {
|
||||
when(streamTracker.addBatchedEventRelay(anyString(), anyString(), anyInt(), anyLong(), any()))
|
||||
.thenReturn(() -> {});
|
||||
|
||||
when(agentService.chat(eq(101L), anyString(), anyString(), any())).thenReturn("Result A");
|
||||
when(agentService.chat(eq(102L), anyString(), anyString(), any())).thenReturn("Result B");
|
||||
when(agentService.chatWithUsage(eq(101L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("Result A"));
|
||||
when(agentService.chatWithUsage(eq(102L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("Result B"));
|
||||
|
||||
String json = "[{\"agentName\":\"AgentA\",\"task\":\"task A\"},{\"agentName\":\"AgentB\",\"task\":\"task B\"}]";
|
||||
|
||||
@ -203,7 +206,8 @@ class DelegateEventSequenceTest {
|
||||
ToolExecutionContext.set("inactive-parent", "admin");
|
||||
when(streamTracker.isRunning("inactive-parent")).thenReturn(false);
|
||||
|
||||
when(agentService.chat(eq(200L), anyString(), anyString(), any())).thenReturn("done");
|
||||
when(agentService.chatWithUsage(eq(200L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("done"));
|
||||
|
||||
// Act
|
||||
delegateAgentTool.delegateToAgent("QuietAgent", "quiet task", null, null);
|
||||
@ -235,7 +239,7 @@ class DelegateEventSequenceTest {
|
||||
return (Runnable) () -> {};
|
||||
});
|
||||
|
||||
when(agentService.chat(eq(300L), anyString(), anyString(), any()))
|
||||
when(agentService.chatWithUsage(eq(300L), anyString(), anyString(), any()))
|
||||
.thenAnswer(invocation -> {
|
||||
BiConsumer<String, String> relay = relayRef.get();
|
||||
// These should produce delegation_progress:
|
||||
@ -244,7 +248,7 @@ class DelegateEventSequenceTest {
|
||||
// These should be ignored by the relay filter:
|
||||
relay.accept("heartbeat", "{}");
|
||||
relay.accept("token", "{\"text\":\"hello\"}");
|
||||
return "filtered result";
|
||||
return ChatResult.contentOnly("filtered result");
|
||||
});
|
||||
|
||||
delegateAgentTool.delegateToAgent("FilterAgent", "filter task", null, null);
|
||||
@ -294,18 +298,19 @@ class DelegateEventSequenceTest {
|
||||
// ToolExecutionContext to the Child's own conversation. Reproduce that so
|
||||
// the grandchild's immediate parent resolves to childConv, while its
|
||||
// events must still target rootConv (carried via DelegationContext).
|
||||
when(agentService.chat(eq(100L), anyString(), anyString(), any()))
|
||||
when(agentService.chatWithUsage(eq(100L), anyString(), anyString(), any()))
|
||||
.thenAnswer(inv -> {
|
||||
String childConv = inv.getArgument(2);
|
||||
ToolExecutionContext.set(childConv, "admin");
|
||||
try {
|
||||
return delegateAgentTool.delegateToAgent("Grandchild", "gtask", null, null);
|
||||
return ChatResult.contentOnly(
|
||||
delegateAgentTool.delegateToAgent("Grandchild", "gtask", null, null));
|
||||
} finally {
|
||||
ToolExecutionContext.set(rootConv, "admin");
|
||||
}
|
||||
});
|
||||
when(agentService.chat(eq(200L), anyString(), anyString(), any()))
|
||||
.thenReturn("grandchild done");
|
||||
when(agentService.chatWithUsage(eq(200L), anyString(), anyString(), any()))
|
||||
.thenReturn(ChatResult.contentOnly("grandchild done"));
|
||||
|
||||
delegateAgentTool.delegateToAgent("Child", "ctask", null, null);
|
||||
|
||||
|
||||
@ -42,6 +42,13 @@ const progress = computed(() => {
|
||||
return n ? `${n} ${n === 1 ? 'tool' : 'tools'}` : ''
|
||||
})
|
||||
|
||||
// Compact token cost, shown once the subagent finishes (e.g. "3.2k tok").
|
||||
const tokenLabel = computed(() => {
|
||||
const t = (props.node.promptTokens || 0) + (props.node.completionTokens || 0)
|
||||
if (!t) return ''
|
||||
return `${t >= 1000 ? (t / 1000).toFixed(1) + 'k' : t} tok`
|
||||
})
|
||||
|
||||
function stepStatus(i: number): 'pending' | 'running' | 'completed' {
|
||||
const p = plan.value
|
||||
if (!p) return 'pending'
|
||||
@ -63,6 +70,7 @@ function stepStatus(i: number): 'pending' | 'running' | 'completed' {
|
||||
<el-icon class="deleg-node__icon" :size="11"><Connection /></el-icon>
|
||||
<span class="deleg-node__name">{{ node.agentName }}</span>
|
||||
<span v-if="progress" class="deleg-node__badge">{{ progress }}</span>
|
||||
<span v-if="tokenLabel" class="deleg-node__badge deleg-node__badge--token">{{ tokenLabel }}</span>
|
||||
<el-icon v-if="isStalled" class="deleg-node__stale" :title="$t('chat.subagentStalled')" :size="11"><WarningFilled /></el-icon>
|
||||
<el-icon
|
||||
v-if="hasBody"
|
||||
@ -157,6 +165,10 @@ function stepStatus(i: number): 'pending' | 'running' | 'completed' {
|
||||
padding: 0 6px;
|
||||
line-height: 16px;
|
||||
}
|
||||
.deleg-node__badge--token {
|
||||
font-variant-numeric: tabular-nums;
|
||||
opacity: 0.85;
|
||||
}
|
||||
.deleg-node__stale {
|
||||
flex-shrink: 0;
|
||||
color: var(--mc-warning, #e6a23c);
|
||||
|
||||
@ -1072,14 +1072,25 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
}
|
||||
|
||||
/** Mark a subagent (segment or nested node) complete by subagentId. */
|
||||
// Compact "(12s · 3.2k tok)" meta suffix for a completed delegation segment.
|
||||
function delegMetaSuffix(durationMs?: number, promptTokens?: number, completionTokens?: number): string {
|
||||
const parts: string[] = []
|
||||
if (durationMs) parts.push(`${Math.round(durationMs / 1000)}s`)
|
||||
const tok = (promptTokens || 0) + (completionTokens || 0)
|
||||
if (tok > 0) parts.push(`${tok >= 1000 ? (tok / 1000).toFixed(1) + 'k' : tok} tok`)
|
||||
return parts.length ? ` (${parts.join(' · ')})` : ''
|
||||
}
|
||||
|
||||
function markDelegComplete(segs: MessageSegment[], subagentId: string | undefined, childConvId: string | undefined,
|
||||
success: boolean, resultPreview?: string, durationMs?: number): boolean {
|
||||
success: boolean, resultPreview?: string, durationMs?: number,
|
||||
promptTokens?: number, completionTokens?: number): boolean {
|
||||
const seg = findDelegSegment(segs, subagentId, childConvId)
|
||||
if (seg) {
|
||||
seg.status = success ? 'completed' : 'error'
|
||||
seg.toolSuccess = success
|
||||
if (resultPreview) seg.toolResult = resultPreview
|
||||
if (durationMs) seg.toolArgs = (seg.toolArgs || '').trimEnd() + ` (${Math.round(durationMs / 1000)}s)`
|
||||
const suffix = delegMetaSuffix(durationMs, promptTokens, completionTokens)
|
||||
if (suffix) seg.toolArgs = (seg.toolArgs || '').trimEnd() + suffix
|
||||
return true
|
||||
}
|
||||
if (subagentId) {
|
||||
@ -1089,6 +1100,8 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
node.status = success ? 'completed' : 'error'
|
||||
if (resultPreview) node.result = resultPreview
|
||||
if (durationMs) node.durationMs = durationMs
|
||||
if (promptTokens) node.promptTokens = promptTokens
|
||||
if (completionTokens) node.completionTokens = completionTokens
|
||||
return true
|
||||
}
|
||||
}
|
||||
@ -1232,7 +1245,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
if (isStaleEvent(data)) return
|
||||
if (!currentAssistantId.value) return
|
||||
markDelegComplete(currentSegments.value, data.subagentId, data.childConversationId,
|
||||
!!data.success, data.resultPreview, data.durationMs)
|
||||
!!data.success, data.resultPreview, data.durationMs, data.promptTokens, data.completionTokens)
|
||||
flushSegmentsToMessage()
|
||||
})
|
||||
|
||||
@ -1250,7 +1263,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
: !!(cr.subagentId && findNode(segs.flatMap(s => s.childTimeline?.children || []), cr.subagentId)?.status === 'running')
|
||||
if (stillRunning) {
|
||||
markDelegComplete(segs, cr.subagentId, cr.childConversationId, !!cr.success,
|
||||
cr.error || undefined, cr.durationMs)
|
||||
cr.error || undefined, cr.durationMs, cr.promptTokens, cr.completionTokens)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
@ -1261,7 +1274,7 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
}
|
||||
} else {
|
||||
markDelegComplete(segs, data.subagentId, data.childConversationId,
|
||||
!!data.success, data.resultPreview, data.durationMs)
|
||||
!!data.success, data.resultPreview, data.durationMs, data.promptTokens, data.completionTokens)
|
||||
}
|
||||
flushSegmentsToMessage()
|
||||
})
|
||||
|
||||
@ -171,6 +171,9 @@ export interface DelegationNode {
|
||||
tools?: DelegationToolEntry[]
|
||||
result?: string
|
||||
durationMs?: number
|
||||
/** Child token usage relayed from delegation_end / delegation_child_complete. */
|
||||
promptTokens?: number
|
||||
completionTokens?: number
|
||||
/** Heartbeat watchdog flagged this subagent as making no observable progress. */
|
||||
stale?: boolean
|
||||
/** Spawned via fire-and-forget delegation: runs detached, result via task_output. */
|
||||
|
||||
Loading…
Reference in New Issue
Block a user