mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-15 03:55:09 +08:00
fix(agent): unblock multi-role parallel delegation + propagate force-stop
This commit is contained in:
parent
84ddc6b46a
commit
2e0ff8f90f
@ -74,6 +74,20 @@ public class ConversationWindowManager {
|
|||||||
private static final int CONTENT_TAIL = 1500;
|
private static final int CONTENT_TAIL = 1500;
|
||||||
private static final int OLD_TOOL_RESULT_SUMMARY_THRESHOLD = 500;
|
private static final int OLD_TOOL_RESULT_SUMMARY_THRESHOLD = 500;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Tool names whose results must never be compacted into a one-line
|
||||||
|
* summary. Sub-agent delegations are irreplaceable: the child runs an
|
||||||
|
* independent LLM session that the parent cannot reproduce, so dropping
|
||||||
|
* earlier batches forces the parent to re-dispatch the same children to
|
||||||
|
* recover what was lost. Every other tool (read_file, shell, search,
|
||||||
|
* memory) can be re-invoked cheaply if the parent decides it needs
|
||||||
|
* the data again.
|
||||||
|
*/
|
||||||
|
private static final java.util.Set<String> PRUNE_EXEMPT_TOOLS = java.util.Set.of(
|
||||||
|
"delegateToAgent",
|
||||||
|
"delegateParallel"
|
||||||
|
);
|
||||||
|
|
||||||
// ==================== 冷却机制 ====================
|
// ==================== 冷却机制 ====================
|
||||||
|
|
||||||
/** 摘要失败后的冷却时间(毫秒):10 分钟 */
|
/** 摘要失败后的冷却时间(毫秒):10 分钟 */
|
||||||
@ -384,7 +398,8 @@ public class ConversationWindowManager {
|
|||||||
boolean messageChanged = false;
|
boolean messageChanged = false;
|
||||||
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
for (ToolResponseMessage.ToolResponse r : trm.getResponses()) {
|
||||||
String data = r.responseData();
|
String data = r.responseData();
|
||||||
if (keepFull || data == null || data.length() <= OLD_TOOL_RESULT_SUMMARY_THRESHOLD) {
|
boolean exempt = r.name() != null && PRUNE_EXEMPT_TOOLS.contains(r.name());
|
||||||
|
if (keepFull || exempt || data == null || data.length() <= OLD_TOOL_RESULT_SUMMARY_THRESHOLD) {
|
||||||
newResponses.add(r);
|
newResponses.add(r);
|
||||||
if (data != null && data.length() > OLD_TOOL_RESULT_SUMMARY_THRESHOLD) {
|
if (data != null && data.length() > OLD_TOOL_RESULT_SUMMARY_THRESHOLD) {
|
||||||
seenLargeOutputs.add(data);
|
seenLargeOutputs.add(data);
|
||||||
|
|||||||
@ -12,6 +12,7 @@ import vip.mate.audit.service.AuditEventService;
|
|||||||
import vip.mate.channel.web.ChatStreamTracker;
|
import vip.mate.channel.web.ChatStreamTracker;
|
||||||
import vip.mate.common.result.R;
|
import vip.mate.common.result.R;
|
||||||
import vip.mate.exception.MateClawException;
|
import vip.mate.exception.MateClawException;
|
||||||
|
import vip.mate.workspace.conversation.ConversationService;
|
||||||
|
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
@ -35,6 +36,7 @@ public class AgentRuntimeController {
|
|||||||
private final ChatStreamTracker streamTracker;
|
private final ChatStreamTracker streamTracker;
|
||||||
private final SubagentRegistry subagentRegistry;
|
private final SubagentRegistry subagentRegistry;
|
||||||
private final AuditEventService auditEventService;
|
private final AuditEventService auditEventService;
|
||||||
|
private final ConversationService conversationService;
|
||||||
|
|
||||||
@Operation(summary = "Snapshot of every in-flight agent turn")
|
@Operation(summary = "Snapshot of every in-flight agent turn")
|
||||||
@GetMapping("/snapshot")
|
@GetMapping("/snapshot")
|
||||||
@ -59,6 +61,18 @@ public class AgentRuntimeController {
|
|||||||
Authentication auth) {
|
Authentication auth) {
|
||||||
requireAdmin(auth);
|
requireAdmin(auth);
|
||||||
boolean ok = streamTracker.forceRecycle(conversationId);
|
boolean ok = streamTracker.forceRecycle(conversationId);
|
||||||
|
if (ok) {
|
||||||
|
// Flip stream_status off 'running' so the sidebar drops the
|
||||||
|
// 生成中 badge immediately. The downstream doOnCancel /
|
||||||
|
// doOnComplete still fires later (when the agent yields) and may
|
||||||
|
// re-set this to 'idle' — same value, no-op.
|
||||||
|
try {
|
||||||
|
conversationService.updateStreamStatus(conversationId, "idle");
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("recycle: failed to reset stream_status for {}: {}",
|
||||||
|
conversationId, e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
recordAudit(auth, "agent-runtime.recycle", conversationId, Map.of("result", ok));
|
recordAudit(auth, "agent-runtime.recycle", conversationId, Map.of("result", ok));
|
||||||
return R.ok(Map.of("recycled", ok));
|
return R.ok(Map.of("recycled", ok));
|
||||||
}
|
}
|
||||||
@ -89,7 +103,15 @@ public class AgentRuntimeController {
|
|||||||
.toList();
|
.toList();
|
||||||
int recycled = 0;
|
int recycled = 0;
|
||||||
for (String cid : ids) {
|
for (String cid : ids) {
|
||||||
if (streamTracker.forceRecycle(cid)) recycled++;
|
if (streamTracker.forceRecycle(cid)) {
|
||||||
|
recycled++;
|
||||||
|
try {
|
||||||
|
conversationService.updateStreamStatus(cid, "idle");
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("sweep: failed to reset stream_status for {}: {}",
|
||||||
|
cid, e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
recordAudit(auth, "agent-runtime.sweep", "all",
|
recordAudit(auth, "agent-runtime.sweep", "all",
|
||||||
Map.of("targets", ids, "recycled", recycled));
|
Map.of("targets", ids, "recycled", recycled));
|
||||||
|
|||||||
@ -1603,6 +1603,18 @@ public class ChatStreamTracker {
|
|||||||
public boolean forceRecycle(String conversationId) {
|
public boolean forceRecycle(String conversationId) {
|
||||||
RunState state = runs.get(conversationId);
|
RunState state = runs.get(conversationId);
|
||||||
if (state == null) return false;
|
if (state == null) return false;
|
||||||
|
// Persist any partial assistant content first — dispose() only severs
|
||||||
|
// the downstream subscription, the agent's worker thread keeps running
|
||||||
|
// and may not yield for minutes. Without this, the conversation row
|
||||||
|
// shows only the user message until the late doOnComplete fires.
|
||||||
|
Runnable callback = state.emergencySaveCallback;
|
||||||
|
if (callback != null) {
|
||||||
|
try {
|
||||||
|
callback.run();
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("forceRecycle: emergency save failed for {}: {}", conversationId, e.getMessage());
|
||||||
|
}
|
||||||
|
}
|
||||||
try {
|
try {
|
||||||
state.stopRequested.set(true);
|
state.stopRequested.set(true);
|
||||||
state.interruptType = InterruptType.USER_STOP;
|
state.interruptType = InterruptType.USER_STOP;
|
||||||
|
|||||||
@ -46,7 +46,16 @@ public class DelegateAgentTool {
|
|||||||
|
|
||||||
private static final int MAX_DELEGATION_DEPTH = 3;
|
private static final int MAX_DELEGATION_DEPTH = 3;
|
||||||
private static final int MAX_RESULT_LENGTH = 4000;
|
private static final int MAX_RESULT_LENGTH = 4000;
|
||||||
private static final int MAX_PARALLEL_CHILDREN = 3;
|
/**
|
||||||
|
* Cap on children dispatched in a single delegateParallel call. Set to 8
|
||||||
|
* because real multi-role evaluations commonly cover 5-8 perspectives
|
||||||
|
* (architecture / backend / frontend / security / cost / ops / contrarian /
|
||||||
|
* progressive-alternative); a lower cap forces the parent to split into
|
||||||
|
* batches, and once an older batch's tool result gets compacted by
|
||||||
|
* ConversationWindowManager the parent can no longer reconstruct what each
|
||||||
|
* child said and starts re-dispatching the same roles in a loop.
|
||||||
|
*/
|
||||||
|
private static final int MAX_PARALLEL_CHILDREN = 8;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* RFC-03 Lane C2 — caps for the parent-context prefix when
|
* RFC-03 Lane C2 — caps for the parent-context prefix when
|
||||||
@ -425,6 +434,14 @@ public class DelegateAgentTool {
|
|||||||
results.add(ChildResult.ofError(idx, agentName, ex.getMessage()));
|
results.add(ChildResult.ofError(idx, agentName, ex.getMessage()));
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
|
// Signal the child's graph to bail out at its next checkpoint.
|
||||||
|
// CompletableFuture.cancel alone only marks the future as
|
||||||
|
// cancelled; without requestStop the underlying ReAct/Plan-Execute
|
||||||
|
// loop keeps invoking LLMs and tools (observed: 8-minute orphan
|
||||||
|
// child still writing files long after the parent gave up).
|
||||||
|
if (p != null && p.childConvId != null) {
|
||||||
|
streamTracker.requestStop(p.childConvId);
|
||||||
|
}
|
||||||
f.cancel(true);
|
f.cancel(true);
|
||||||
// Use ofTimeout so outcome="timeout" is explicit and distinct from "error".
|
// Use ofTimeout so outcome="timeout" is explicit and distinct from "error".
|
||||||
results.add(ChildResult.ofTimeout(idx, agentName, PARALLEL_TIMEOUT_SECONDS));
|
results.add(ChildResult.ofTimeout(idx, agentName, PARALLEL_TIMEOUT_SECONDS));
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user