package loop // 本文件承载**单个/一批工具调用的执行**(span、流式帧、结果归一化、 // 有界并行),与 tools.go 的分类、准入与回渲职责分开 —— 从 tools.go // 抽出,使两边都留在 500 行以内(AGENTS.md §10)。 import ( "context" "fmt" "sync" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace" ) // invokeOne 在其每工具截止时间下执行单个工具调用,发出一个 // tool span、以 tool_execution.start/end 事件流式输出执行进度 // 并归一化结果。执行全程由单一 phase 决策驱动:start 帧带 // running,Invoke 返回后按成败取 done/error,同一取值既决定 // end 帧的 phase,也决定 span 终态(design/19 的 tool 节点 // running/done/error 状态即由此驱动)。 func (l *Loop) invokeOne(ctx context.Context, d *tool.Dispatcher, tc agent.ToolCall) ToolOutcome { // speak 由循环自己承载(见 invokeTool),没有 dispatcher 也能执行;其余 // 工具缺 dispatcher 时仍按原样短路,不为此多开 span 或流式帧。 if d == nil && !(l.voiceEnabled() && tc.Name == speakToolName) { return ToolOutcome{Err: fmt.Errorf("no tool dispatcher configured for %q", tc.Name), ErrKind: tool.KindExecution} } _, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool) l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionStart, model.ToolPhaseRunning, "") // 分派见 invokeTool:路标识桥(tool.WithLegID)与 speak 的循环内承载都在 // 那里。路标识(design/15 §2.5.7、主人裁决 3):网关在**每一条在途路**上把 // 互斥身份经 loop.WithToolLegID 挂在 ctx 上(普通回合 / 并行路 / 队首续跑路 // 都挂,取值 = 该路的 turnID),工具层凭 tool.LegIDFrom 读到「我是哪一路」 // 做文件级互斥;未挂载 ⇒ 空串 ⇒ 完全不做互斥,行为逐字不变(兼容硬要求)。 // // 刻意**不**读 ParallelLegFrom:那条通道只有真并行路才有、且只决定 checkpoint // 归属(checkpoint.go 的 checkpointAgentID),拿它当互斥身份会让普通回合在 // 「1 普通回合 + 1 并行路」时完全不受保护。 result, err := l.invokeTool(ctx, d, tc) phase := model.ToolPhaseDone if err != nil { phase = model.ToolPhaseError } l.emitToolExecution(ctx, tc, model.StreamKindToolExecutionEnd, phase, toolEndFrameSummary(result, err)) out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)} l.spanProp(toolSpan, "phase", phase) l.recordToolDone(toolSpan, tc, out) if err != nil { l.spanProp(toolSpan, "error", err.Error()) l.spanProp(toolSpan, "error_kind", out.ErrKind) } else { recordToolResult(toolSpan, result) } l.finishSpan(toolSpan, spanStatusForToolPhase(phase)) return out } // emitToolExecution 在循环处于流式路径(配置了 OnStream)时把 // tool_execution 三事件(start / update / end)转发给前端;否则为 // 空操作(被静音的内部流程也会跳过它)。phase 是结构化字段 // (running/done/error),绝不塞进 Delta;工具名走 Tool 字段, // ID 复用 provider 的 tool call id,使执行帧能与其 tool_call.* 帧关联; // resultSummary 只在 end 帧、且只在结果真正拿到时非空(口径见 tool_frame_summary.go)。 func (l *Loop) emitToolExecution(ctx context.Context, tc agent.ToolCall, kind model.StreamKind, phase, resultSummary string) { if l.onStream == nil || isMutedBroadcast(ctx) { return } ev := model.StreamEvent{Kind: kind, Phase: phase, Tool: tc.Name, ID: tc.ID, ResultSummary: resultSummary} ev.SessionID = loopSession(ctx) ev.TurnID = loopTurn(ctx) l.onStream(ev) } // spanStatusForToolPhase 把工具执行的终态 phase 映射为 span 终态, // 使 trace/流程图与 WS 帧从同一个 phase 取值得出一致结论。 // 非终态 phase(running)不会走到这里 —— span 只在执行结束时收尾。 func spanStatusForToolPhase(phase string) trace.SpanStatus { if phase == model.ToolPhaseError { return trace.SpanStatusError } return trace.SpanStatusOK } // invokeParallel 并发运行 idxs 处的调用,上限为循环的 // 并行度,并把每个结果写回 out 的对应调用索引处,使调用方 // 能按原始顺序渲染结果。信号量限制并发,在途 // goroutine 完成时释放其槽位,因此被取消的 context // (它使每个 Invoke 迅速返回)不会泄露 goroutine。 func (l *Loop) invokeParallel(ctx context.Context, d *tool.Dispatcher, calls []agent.ToolCall, idxs []int, out []ToolOutcome) { if len(idxs) == 0 { return } sem := make(chan struct{}, l.toolParallelism) var wg sync.WaitGroup for _, i := range idxs { wg.Add(1) sem <- struct{}{} go func(i int) { defer wg.Done() defer func() { <-sem }() out[i] = l.invokeOne(ctx, d, calls[i]) }(i) } wg.Wait() }