mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 03:13:41 +08:00
fix(delegate): fix parallel timeout + add real-time per-child visibility
This commit is contained in:
matevip
parent
95dd500a16
commit
2e15369465
@ -40,7 +40,13 @@ public class DelegateAgentTool {
|
||||
private static final int MAX_DELEGATION_DEPTH = 3;
|
||||
private static final int MAX_RESULT_LENGTH = 4000;
|
||||
private static final int MAX_PARALLEL_CHILDREN = 3;
|
||||
private static final int PARALLEL_TIMEOUT_SECONDS = 60;
|
||||
/**
|
||||
* Per-child timeout — raised from 60 s to 120 s so that slow LLM models
|
||||
* (kimi-code observed p99 ≈ 91 s) can complete before the parent gives up.
|
||||
* The previous 60 s limit was structurally impossible to satisfy once any
|
||||
* child called an LLM-backed tool.
|
||||
*/
|
||||
private static final int PARALLEL_TIMEOUT_SECONDS = 120;
|
||||
|
||||
/** 子 Agent 禁用的工具：防递归 + 防副作用 */
|
||||
private static final Set<String> CHILD_DENIED_TOOLS = Set.of(
|
||||
@ -203,6 +209,26 @@ public class DelegateAgentTool {
|
||||
CompletableFuture<ChildResult> future = CompletableFuture.supplyAsync(
|
||||
() -> runSingleChild(p.index, p.agent, p.task, parentConversationId, p.childConvId),
|
||||
DELEGATION_EXECUTOR);
|
||||
|
||||
// Broadcast per-child completion as soon as each child finishes
|
||||
// — frontend can update that child's status without waiting for all children.
|
||||
if (hasParent) {
|
||||
final String parentConvIdFinal = parentConversationId;
|
||||
future.whenComplete((result, ex) -> {
|
||||
if (!streamTracker.isRunning(parentConvIdFinal)) return;
|
||||
ChildResult r = (result != null) ? result
|
||||
: ChildResult.error(p.index, p.agent.getName(), ex != null ? ex.getMessage() : "Unknown error");
|
||||
streamTracker.broadcastObject(parentConvIdFinal, "delegation_child_complete", Map.of(
|
||||
"taskIndex", r.taskIndex,
|
||||
"childConversationId", p.childConvId,
|
||||
"childAgentName", r.agentName,
|
||||
"success", r.success,
|
||||
"durationMs", r.durationMs,
|
||||
"resultPreview", r.success ? truncate(r.result, 150)
|
||||
: (r.error != null ? r.error : "error")));
|
||||
});
|
||||
}
|
||||
|
||||
futures.put(p.index, future);
|
||||
}
|
||||
|
||||
@ -243,14 +269,24 @@ public class DelegateAgentTool {
|
||||
if (p.stopRelay != null) p.stopRelay.run();
|
||||
}
|
||||
|
||||
// 7. 广播 delegation_end
|
||||
// 7. 广播 delegation_end（含每个子任务的摘要，前端可用于展示分项结果）
|
||||
if (hasParent) {
|
||||
List<Map<String, Object>> childResults = results.stream().map(r -> {
|
||||
Map<String, Object> m = new java.util.LinkedHashMap<>();
|
||||
m.put("taskIndex", r.taskIndex);
|
||||
m.put("agentName", r.agentName);
|
||||
m.put("success", r.success);
|
||||
m.put("durationMs", r.durationMs);
|
||||
if (!r.success && r.error != null) m.put("error", r.error);
|
||||
return m;
|
||||
}).toList();
|
||||
streamTracker.broadcastObject(parentConversationId, "delegation_end", Map.of(
|
||||
"parallel", true,
|
||||
"totalDurationMs", totalDurationMs,
|
||||
"success", results.stream().allMatch(r -> r.success),
|
||||
"completedCount", results.stream().filter(r -> r.success).count(),
|
||||
"totalCount", results.size()));
|
||||
"totalCount", results.size(),
|
||||
"childResults", childResults));
|
||||
}
|
||||
|
||||
// 8. 构建返回结果
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
<script setup lang="ts">
|
||||
import { ref, computed, watch } from 'vue'
|
||||
import { Loading, Select, CloseBold, ArrowDown, Document, Setting } from '@element-plus/icons-vue'
|
||||
import { Loading, Select, CloseBold, ArrowDown, Document, Setting, Connection } from '@element-plus/icons-vue'
|
||||
import { useToolLabel } from '@/composables/useToolLabel'
|
||||
import type { MessageSegment } from '@/types'
|
||||
|
||||
@ -17,7 +17,16 @@ watch(() => props.segment.status, (val) => {
|
||||
if (val !== 'running') expanded.value = false
|
||||
})
|
||||
|
||||
const displayName = computed(() => getToolLabel(props.segment.toolName || ''))
|
||||
/** Delegation segments are identified by the → prefix injected in useChat.ts. */
|
||||
const isDelegation = computed(() => (props.segment.toolName || '').startsWith('→'))
|
||||
|
||||
const displayName = computed(() => {
|
||||
const raw = props.segment.toolName || ''
|
||||
// Strip the → prefix for delegation segments so getToolLabel works cleanly,
|
||||
// then prepend it back as a visual indicator.
|
||||
if (isDelegation.value) return `→ ${getToolLabel(raw.slice(1).trim())}`
|
||||
return getToolLabel(raw)
|
||||
})
|
||||
|
||||
const truncatedArgs = computed(() => {
|
||||
const args = props.segment.toolArgs || ''
|
||||
@ -28,7 +37,9 @@ const truncatedArgs = computed(() => {
|
||||
const s = vals.join(', ')
|
||||
return s.length > 80 ? s.slice(0, 80) + '...' : s
|
||||
} catch {
|
||||
return args.length > 80 ? args.slice(0, 80) + '...' : args
|
||||
// Multi-line delegation progress — show first line only in collapsed view
|
||||
const firstLine = args.split('\n')[0].trim()
|
||||
return firstLine.length > 80 ? firstLine.slice(0, 80) + '...' : firstLine
|
||||
}
|
||||
})
|
||||
|
||||
@ -56,7 +67,8 @@ const isRunning = computed(() => props.segment.status === 'running')
|
||||
<el-icon v-else :size="13"><CloseBold /></el-icon>
|
||||
</span>
|
||||
<span class="seg-tool__type-icon">
|
||||
<el-icon v-if="isRead" :size="12"><Document /></el-icon>
|
||||
<el-icon v-if="isDelegation" :size="12"><Connection /></el-icon>
|
||||
<el-icon v-else-if="isRead" :size="12"><Document /></el-icon>
|
||||
<el-icon v-else :size="12"><Setting /></el-icon>
|
||||
</span>
|
||||
<span class="seg-tool__name">{{ displayName }}</span>
|
||||
|
||||
@ -636,18 +636,76 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
|
||||
stream.on('delegation_progress', (data) => {
|
||||
if (isStaleEvent(data)) return
|
||||
if (currentAssistantId.value && data.originalEvent === 'tool_call_started') {
|
||||
const segs = currentSegments.value
|
||||
// 按 childAgentName 匹配对应的 delegation segment（并行时多个）
|
||||
const childName = data.childAgentName || ''
|
||||
const delegSeg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`)
|
||||
|| segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
if (delegSeg) {
|
||||
const childData = typeof data.data === 'string' ? data.data : JSON.stringify(data.data)
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '') + '\n [子任务] ' + childData
|
||||
if (!currentAssistantId.value) return
|
||||
const segs = currentSegments.value
|
||||
const childName = data.childAgentName || ''
|
||||
// Find the running delegation segment for this child (or fall back to any running delegation)
|
||||
const delegSeg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`)
|
||||
|| segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
|
||||
if (!delegSeg) return
|
||||
|
||||
if (data.originalEvent === 'tool_call_started') {
|
||||
// Child started a sub-tool — append activity hint so the user sees the child is working
|
||||
const childData = data.data
|
||||
const toolName = typeof childData === 'object' ? childData?.toolName : String(childData || '')
|
||||
if (toolName) {
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '') + `\n → ${toolName}`
|
||||
}
|
||||
} else if (data.originalEvent === 'tool_call_completed') {
|
||||
// Child finished a sub-tool call — update the running hint
|
||||
const childData = data.data
|
||||
const toolName = typeof childData === 'object' ? childData?.toolName : String(childData || '')
|
||||
const success = typeof childData === 'object' ? childData?.success !== false : true
|
||||
if (toolName) {
|
||||
// Replace last appended "→ toolName" with "✓/✗ toolName"
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '').replace(
|
||||
new RegExp(`\\n → ${toolName.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}\\s*$`),
|
||||
`\n ${success ? '✓' : '✗'} ${toolName}`)
|
||||
}
|
||||
} else if (data.originalEvent === 'phase') {
|
||||
// Child entered a new phase (reasoning, executing_tool, etc.)
|
||||
const phase = typeof data.data === 'object' ? data.data?.phase : String(data.data || '')
|
||||
const phaseHints: Record<string, string> = {
|
||||
reasoning: '…',
|
||||
executing_tool: '→',
|
||||
planning: '📋',
|
||||
summarizing: '✍',
|
||||
}
|
||||
const hint = phaseHints[phase]
|
||||
if (hint && !delegSeg.toolArgs?.endsWith(hint)) {
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '').trimEnd() + ' ' + hint
|
||||
}
|
||||
}
|
||||
flushSegmentsToMessage()
|
||||
})
|
||||
|
||||
// Per-child completion: fires as soon as each individual child agent finishes,
|
||||
// before the overall delegation_end. Marks that child's segment done immediately
|
||||
// so the user sees incremental progress rather than a bulk update at the end.
|
||||
stream.on('delegation_child_complete', (data) => {
|
||||
if (isStaleEvent(data)) return
|
||||
if (!currentAssistantId.value) return
|
||||
const segs = currentSegments.value
|
||||
const childName = data.childAgentName || ''
|
||||
const delegSeg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName === `→ ${childName}`)
|
||||
|| segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
if (delegSeg) {
|
||||
delegSeg.status = data.success ? 'completed' : 'error'
|
||||
delegSeg.toolSuccess = data.success
|
||||
// Append duration to args so the user sees how long each child took
|
||||
if (data.durationMs) {
|
||||
const durSec = Math.round(data.durationMs / 1000)
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '').trimEnd() + ` (${durSec}s)`
|
||||
}
|
||||
if (!data.success && data.resultPreview) {
|
||||
delegSeg.toolResult = data.resultPreview
|
||||
}
|
||||
}
|
||||
flushSegmentsToMessage()
|
||||
})
|
||||
|
||||
stream.on('delegation_end', (data) => {
|
||||
@ -655,21 +713,42 @@ export function useChat(options: UseChatOptions): UseChatReturn {
|
||||
if (currentAssistantId.value) {
|
||||
const segs = currentSegments.value
|
||||
if (data.parallel) {
|
||||
// 并行模式：关闭所有 running 的 delegation segments
|
||||
const totalMs = data.totalDurationMs ? Math.round(data.totalDurationMs / 1000) : 0
|
||||
segs.filter((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
.forEach((s: MessageSegment) => {
|
||||
s.status = 'completed'
|
||||
s.toolName = (s.toolName || '') + (data.success ? ' ✓' : ' ✗')
|
||||
})
|
||||
// Parallel mode: use per-child results if available (new backend),
|
||||
// fall back to aggregate success flag for older backends.
|
||||
if (Array.isArray(data.childResults) && data.childResults.length > 0) {
|
||||
for (const cr of data.childResults) {
|
||||
const agentName = cr.agentName || ''
|
||||
const seg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' &&
|
||||
(s.status === 'running' || s.status === 'completed') &&
|
||||
s.toolName?.includes(agentName))
|
||||
if (seg && seg.status === 'running') {
|
||||
// Segment not yet closed by delegation_child_complete (e.g. timed out child)
|
||||
seg.status = cr.success ? 'completed' : 'error'
|
||||
seg.toolSuccess = cr.success
|
||||
if (cr.durationMs) {
|
||||
const durSec = Math.round(cr.durationMs / 1000)
|
||||
seg.toolArgs = (seg.toolArgs || '').trimEnd() + ` (${durSec}s)`
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Legacy fallback: mark all remaining running delegation segments with overall status
|
||||
segs.filter((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
.forEach((s: MessageSegment) => {
|
||||
s.status = data.success ? 'completed' : 'error'
|
||||
})
|
||||
}
|
||||
} else {
|
||||
// 单任务模式
|
||||
const delegSeg = segs.findLast((s: MessageSegment) => s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
// Single-task mode
|
||||
const delegSeg = segs.findLast((s: MessageSegment) =>
|
||||
s.type === 'tool_call' && s.status === 'running' && s.toolName?.startsWith('→'))
|
||||
if (delegSeg) {
|
||||
delegSeg.status = 'completed'
|
||||
delegSeg.toolName = (delegSeg.toolName || '') + (data.success ? ' ✓' : ' ✗')
|
||||
delegSeg.status = data.success ? 'completed' : 'error'
|
||||
delegSeg.toolSuccess = data.success
|
||||
if (data.durationMs) {
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '') + `\n 耗时: ${Math.round(data.durationMs / 1000)}s`
|
||||
delegSeg.toolArgs = (delegSeg.toolArgs || '').trimEnd() + ` (${Math.round(data.durationMs / 1000)}s)`
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -43,6 +43,7 @@ export type SSEEventType =
|
||||
| 'delegation_start'
|
||||
| 'delegation_progress'
|
||||
| 'delegation_end'
|
||||
| 'delegation_child_complete'
|
||||
|
||||
export interface SSEEvent {
|
||||
type: SSEEventType
|
||||
|
||||
Loading…
Reference in New Issue
Block a user