feat(loop): wire speak tool into the loop (M57)
- speak_tool.go: the speak tool itself — rune-derived timeout budget, structured failure codes, persona/emotion stamped on the synthesis ctx, audio persisted through the store seam, structured result for the model. - voice_deps.go: the wiring seam (VoiceSynth / VoiceStore + VoiceStoreFunc adapter), voiceEnabled, withSpeakSpec (appended after every schema shrink), and captureVoice (success beats failure, last one wins). - options.go / loop.go: VoiceSynth, VoiceStore, SpeakBudget injection points. - request.go: advertise speak on both toolSchemas return paths. - tools.go / tools_invoke.go: carry the batch voice outcome; split invokeOne / emitToolExecution / spanStatusForToolPhase / invokeParallel into their own file (tools.go was at the 500-line ceiling) and let speak run without a dispatcher. - run.go: keep the voice outcome across tool batches (a later batch without speak must not erase it). Known, deliberate: speak bypasses the capacity allow-set and skill-round narrowing (it belongs to no built-in toolset), so it is also advertised in quiet/awake; execution still passes the tool security gate, where speak is not registered and non-owner actors are denied fail-closed.
This commit is contained in:
AskaEth committed
1 parent
f84a358e30
commit
1347a77d33
8 files changed
+628
-94
No files matched your search
@@ -5,12 +5,10 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"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"
|
||||
)
|
||||
|
||||
// Options.ToolExecution 接受的 ToolExecutionMode 取值。为空时选用
|
||||
@@ -77,6 +75,9 @@ type toolRunResult struct {
|
||||
messages []model.Message
|
||||
executed int
|
||||
terminate bool
|
||||
// voice 是本批 speak 调用的结论（M57）：nil = 这一批没有要求语音。
|
||||
// 由 runToolCalls 在结局钩子之后提取，供 run.go 回填 Result.Voice。
|
||||
voice *OutboundVoice
|
||||
}
|
||||
|
||||
// toolLoopState 携带关键工具循环安全网的每次 run 状态。
|
||||
@@ -156,6 +157,10 @@ func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls
|
||||
// 结果落定后的两类观察/改写钩子见 applyToolOutcomeHooks。
|
||||
l.applyToolOutcomeHooks(ctx, calls, outcomes)
|
||||
|
||||
// 语音结论必须在钩子之后提取：AfterOutcome 可以改写 speak 的结果，
|
||||
// 出站结论要看**模型最终看到的那份**。
|
||||
res.voice = l.captureVoice(calls, outcomes)
|
||||
|
||||
// 钩子可以把每个调用都标记为 terminate 来要求终止本轮
|
||||
// （OpenClaw 的 shouldTerminateToolBatch）。
|
||||
if !res.terminate && shouldTerminateBatch(outcomes) {
|
||||
@@ -287,95 +292,6 @@ func (l *Loop) applyToolOutcomeHooks(ctx context.Context, calls []agent.ToolCall
|
||||
}
|
||||
}
|
||||
|
||||
// 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 {
|
||||
if d == nil {
|
||||
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, "")
|
||||
// 路标识桥（design/15 §2.5.7、主人裁决 3）：网关在**每一条在途路**上把互斥
|
||||
// 身份经 loop.WithToolLegID 挂在 ctx 上（普通回合 / 并行路 / 队首续跑路都挂，
|
||||
// 取值 = 该路的 turnID），这里把它桥接成工具层的 tool.WithLegID —— 工具侧
|
||||
// 凭 tool.LegIDFrom 读到「我是哪一路」做文件级互斥；未挂载 ⇒ 空串 ⇒ 完全
|
||||
// 不做互斥，行为逐字不变（兼容硬要求）。
|
||||
//
|
||||
// 刻意**不**读 ParallelLegFrom：那条通道只有真并行路才有、且只决定 checkpoint
|
||||
// 归属（checkpoint.go 的 checkpointAgentID），拿它当互斥身份会让普通回合在
|
||||
// 「1 普通回合 + 1 并行路」时完全不受保护。
|
||||
result, err := d.Invoke(tool.WithLegID(ctx, ToolLegIDFrom(ctx)), tc.Name, tc.Args)
|
||||
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()
|
||||
}
|
||||
|
||||
// 由 classifyToolBatch 为两种跳过原因返回。
|
||||
func duplicateCallReason() string {
|
||||
return "duplicate call with identical arguments already issued in this batch"
|
||||
|
||||
Reference in new issue
Block a user