feat(loop): Top5#4 工具循环增强(对齐 OpenClaw agent-loop)——重复检测/并行串行/钩子/errorKind/repair
- 重复调用安全网(critical-tool-loop):指纹=name+canonical JSON(args);批内重复第二次跳过回喂
「duplicate call」;跨轮关键重复(同指纹重现)→首次 block+criticalSeen,再次连续重复→终止整轮
StopReasonToolLoopTerminated(替代裸 8 轮针对同调用的硬截止)。
- 并行/串行混合:invokeParallel 默认并行,信号量(ToolParallelism 默认4)限并发,结果按调用索引
写回按原序渲染;串行触发=Options.ToolExecution:sequential 或 SetToolExecutionMode(name,sequential)
(对齐 OpenClaw「一个 sequential 标记→整组串行」);goroutine 不泄漏、ctx 取消快速返回。
- 工具钩子:ToolLoopHooks(BatchAdmit/CallAdmit/AfterCall/AfterOutcome) 经 Options.Hooks 注入,
均 nil-tolerant;供后续敏感操作代审复用(design/11)。terminate 用 every 语义。
- status 分类:ToolError.Kind() 区分 argument-validation(JSON-Schema 失败)/execution;
失败消息渲染 {kind,message} 让模型区分重发参数 vs 执行失败。
- tool-call-repair:LenientJSONUnmarshal 修 fence 包裹/尾逗号畸形;agent.extractToolCalls 与
loop.decodeToolCalls 走它;不可修复退化为文本不崩整轮。意图调用未区分回喂提示(后置)。
- tool_execution_update:流式路径 emitToolExecution 发 StreamKindToolExecution(start/end/end:error)。
go build/vet/test 全绿(46包)。TestRun_MaxRounds 改每轮不同参(原同 echo 被安全网拦,预期新行为)。
-race 受宿主 TSan VMA 限制无法运行(与代码无关)。
This commit is contained in:
1 parent
2248ddce15
commit
35cb2d3942
11 files changed
+1079
-74
No files matched your search
@@ -2,11 +2,11 @@ package agent
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/tool"
|
||||
)
|
||||
|
||||
// Stop reasons carried by StepResult.StopReason, consumed by the loop (M2) to
|
||||
@@ -119,7 +119,11 @@ func extractToolCalls(resp model.Response) ([]ToolCall, bool) {
|
||||
var payload struct {
|
||||
ToolCalls []ToolCall `json:"tool_calls"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(resp.Content), &payload); err != nil {
|
||||
// Lenient decode repairs common model-emitted malformations (code fences,
|
||||
// trailing commas) so a slightly broken tool-call envelope does not silently
|
||||
// turn into a text answer; an unrecoverable payload still yields ok=false and
|
||||
// the round proceeds as text (the loop never crashes on a bad call).
|
||||
if err := tool.LenientJSONUnmarshal([]byte(resp.Content), &payload); err != nil {
|
||||
return nil, false
|
||||
}
|
||||
if len(payload.ToolCalls) == 0 {
|
||||
|
||||
@@ -36,6 +36,15 @@ const (
|
||||
// StopReasonMaxRounds reports that the loop exhausted its round budget
|
||||
// before reaching a final answer.
|
||||
StopReasonMaxRounds = "max_rounds"
|
||||
|
||||
// StopReasonToolLoopTerminated reports that the critical-tool-loop safety net
|
||||
// stopped the run after the same tool call was repeated identically across
|
||||
// consecutive rounds.
|
||||
StopReasonToolLoopTerminated = "tool_loop_terminated"
|
||||
|
||||
// DefaultToolParallelism caps how many tool calls run concurrently when the
|
||||
// loop runs a batch in parallel mode and no explicit cap is configured.
|
||||
DefaultToolParallelism = 4
|
||||
)
|
||||
|
||||
// Sentinel errors returned by the Loop. Use errors.Is to test for them; they are
|
||||
@@ -69,19 +78,22 @@ type Result struct {
|
||||
// dispatcher, stores, recorder, tracer) are themselves concurrency-safe, which
|
||||
// they are by contract. Run keeps all loop state on the stack.
|
||||
type Loop struct {
|
||||
agent *agent.Agent
|
||||
tools *tool.Dispatcher
|
||||
memory *memory.Store
|
||||
retriever *memory.Retriever
|
||||
working *memory.WorkingMemory
|
||||
extractor memory.Extractor
|
||||
checkpoint *persist.Store
|
||||
tracer *trace.Tracer
|
||||
guards *guardrail.Pipeline
|
||||
personaID string
|
||||
maxRounds int
|
||||
onStream func(model.StreamEvent)
|
||||
log *slog.Logger
|
||||
agent *agent.Agent
|
||||
tools *tool.Dispatcher
|
||||
memory *memory.Store
|
||||
retriever *memory.Retriever
|
||||
working *memory.WorkingMemory
|
||||
extractor memory.Extractor
|
||||
checkpoint *persist.Store
|
||||
tracer *trace.Tracer
|
||||
guards *guardrail.Pipeline
|
||||
personaID string
|
||||
maxRounds int
|
||||
toolExec string
|
||||
toolParallelism int
|
||||
hooks ToolLoopHooks
|
||||
onStream func(model.StreamEvent)
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
// New builds a Loop from opts, applying defaults. A non-positive MaxRounds
|
||||
@@ -94,20 +106,31 @@ func New(opts Options) *Loop {
|
||||
if maxRounds <= 0 {
|
||||
maxRounds = DefaultMaxRounds
|
||||
}
|
||||
toolExec := opts.ToolExecution
|
||||
if toolExec == "" {
|
||||
toolExec = toolExecParallel
|
||||
}
|
||||
toolParallelism := opts.ToolParallelism
|
||||
if toolParallelism <= 0 {
|
||||
toolParallelism = DefaultToolParallelism
|
||||
}
|
||||
return &Loop{
|
||||
agent: opts.Agent,
|
||||
tools: opts.Tools,
|
||||
memory: opts.Memory,
|
||||
retriever: opts.Retriever,
|
||||
working: opts.WorkingMemory,
|
||||
extractor: opts.Extractor,
|
||||
checkpoint: opts.Checkpoint,
|
||||
tracer: opts.Tracer,
|
||||
guards: opts.Guards,
|
||||
personaID: opts.PersonaID,
|
||||
maxRounds: maxRounds,
|
||||
onStream: opts.OnStream,
|
||||
log: sourcedLog(opts.Log),
|
||||
agent: opts.Agent,
|
||||
tools: opts.Tools,
|
||||
memory: opts.Memory,
|
||||
retriever: opts.Retriever,
|
||||
working: opts.WorkingMemory,
|
||||
extractor: opts.Extractor,
|
||||
checkpoint: opts.Checkpoint,
|
||||
tracer: opts.Tracer,
|
||||
guards: opts.Guards,
|
||||
personaID: opts.PersonaID,
|
||||
maxRounds: maxRounds,
|
||||
toolExec: toolExec,
|
||||
toolParallelism: toolParallelism,
|
||||
hooks: opts.Hooks,
|
||||
onStream: opts.OnStream,
|
||||
log: sourcedLog(opts.Log),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,330 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
)
|
||||
|
||||
func TestRun_DuplicateCallInBatch_Skipped(t *testing.T) {
|
||||
t.Parallel()
|
||||
// Two identical calls in one batch: the second must be skipped, not executed,
|
||||
// and fed back as a duplicate hint.
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}},{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "done"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "done" {
|
||||
t.Fatalf("Text = %q, want done", res.Text)
|
||||
}
|
||||
if res.ToolCallsExecuted != 1 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 1 (duplicate skipped)", res.ToolCallsExecuted)
|
||||
}
|
||||
if res.StopReason != agent.StopReasonStop {
|
||||
t.Fatalf("StopReason = %q, want stop", res.StopReason)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
// Round 1 appended two tool messages: the executed call result, then the
|
||||
// duplicate skip. The last two messages before the final turn are those.
|
||||
result := last[len(last)-2]
|
||||
skip := last[len(last)-1]
|
||||
if !strings.Contains(result.Content[0].Text, "hi") {
|
||||
t.Fatalf("expected echo result, got %q", result.Content[0].Text)
|
||||
}
|
||||
if !strings.Contains(skip.Content[0].Text, "skipped") || !strings.Contains(skip.Content[0].Text, "duplicate") {
|
||||
t.Fatalf("expected duplicate-skip message, got %q", skip.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_CriticalToolLoop_Terminates(t *testing.T) {
|
||||
t.Parallel()
|
||||
// The identical call repeats across rounds: round 2 is blocked (critical
|
||||
// loop flagged), round 3 terminates the run.
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: call}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.StopReason != StopReasonToolLoopTerminated {
|
||||
t.Fatalf("StopReason = %q, want %q", res.StopReason, StopReasonToolLoopTerminated)
|
||||
}
|
||||
if res.ToolCallsExecuted != 1 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 1 (only round 1 executed)", res.ToolCallsExecuted)
|
||||
}
|
||||
if res.Steps != 3 {
|
||||
t.Fatalf("Steps = %d, want 3", res.Steps)
|
||||
}
|
||||
if !strings.Contains(res.Text, "critical tool-loop") {
|
||||
t.Fatalf("Text = %q, want critical tool-loop message", res.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_Parallel_ResultsInCallOrder(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"aa","args":{}},{"name":"bb","args":{}},{"name":"cc","args":{}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "final"}},
|
||||
}}
|
||||
td := dispatcherWith(t,
|
||||
&mockTool{name: "aa", result: "va"},
|
||||
&mockTool{name: "bb", result: "vb"},
|
||||
&mockTool{name: "cc", result: "vc"},
|
||||
)
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8, ToolExecution: "parallel"})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "final" {
|
||||
t.Fatalf("Text = %q, want final", res.Text)
|
||||
}
|
||||
if res.ToolCallsExecuted != 3 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 3", res.ToolCallsExecuted)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
if len(last) < 4 {
|
||||
t.Fatalf("messages = %d, want at least 4 (user + 3 results)", len(last))
|
||||
}
|
||||
got := make([]string, 3)
|
||||
for i := range got {
|
||||
got[i] = last[len(last)-3+i].Content[0].Text
|
||||
}
|
||||
want := []string{
|
||||
"tool aa result: va",
|
||||
"tool bb result: vb",
|
||||
"tool cc result: vc",
|
||||
}
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("result %d = %q, want %q (call order must be preserved)", i, got[i], want[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_Sequential_ExecutesAll(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"aa","args":{}},{"name":"bb","args":{}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "final"}},
|
||||
}}
|
||||
td := dispatcherWith(t,
|
||||
&mockTool{name: "aa", result: "va"},
|
||||
&mockTool{name: "bb", result: "vb"},
|
||||
)
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8, ToolExecution: "sequential"})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.ToolCallsExecuted != 2 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 2", res.ToolCallsExecuted)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ArgumentValidation_ErrorKind(t *testing.T) {
|
||||
t.Parallel()
|
||||
// Missing required field -> the fed-back failure carries errorKind
|
||||
// argument-validation so the model can re-issue a corrected call.
|
||||
call := `{"tool_calls":[{"name":"req","args":{}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "recovered"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "req", params: map[string]any{"required": []any{"must"}}})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.Text != "recovered" {
|
||||
t.Fatalf("Text = %q, want recovered", res.Text)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
lastMsg := last[len(last)-1]
|
||||
if !strings.Contains(lastMsg.Content[0].Text, `"kind":"argument-validation"`) {
|
||||
t.Fatalf("result should carry argument-validation kind, got %q", lastMsg.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_BatchAdmit_BlocksBatch(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "after block"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8, Hooks: ToolLoopHooks{
|
||||
BatchAdmit: func(ctx context.Context, calls []agent.ToolCall) (bool, string) {
|
||||
return false, "policy denied"
|
||||
},
|
||||
}})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.ToolCallsExecuted != 0 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 0 (batch blocked)", res.ToolCallsExecuted)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
lastMsg := last[len(last)-1]
|
||||
if !strings.Contains(lastMsg.Content[0].Text, "policy denied") {
|
||||
t.Fatalf("expected blocked batch message, got %q", lastMsg.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_CallAdmit_VetoesSingleCall(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "after veto"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8, Hooks: ToolLoopHooks{
|
||||
CallAdmit: func(ctx context.Context, call agent.ToolCall) *ToolCallVeto {
|
||||
if call.Name == "echo" {
|
||||
return &ToolCallVeto{Block: true, Reason: "sensitive action requires approval"}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.ToolCallsExecuted != 0 {
|
||||
t.Fatalf("ToolCallsExecuted = %d, want 0 (call vetoed)", res.ToolCallsExecuted)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
lastMsg := last[len(last)-1]
|
||||
if !strings.Contains(lastMsg.Content[0].Text, "sensitive action requires approval") {
|
||||
t.Fatalf("expected veto message, got %q", lastMsg.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_AfterCall_RewritesResult(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "final"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "raw"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8, Hooks: ToolLoopHooks{
|
||||
AfterCall: func(ctx context.Context, call agent.ToolCall, outcome ToolOutcome) *ToolOutcome {
|
||||
if call.Name == "echo" {
|
||||
outcome.Result = "rewritten"
|
||||
return &outcome
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
last := ad.reqs[len(ad.reqs)-1].Messages
|
||||
lastMsg := last[len(last)-1]
|
||||
if !strings.Contains(lastMsg.Content[0].Text, "rewritten") {
|
||||
t.Fatalf("expected rewritten result, got %q", lastMsg.Content[0].Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_ToolCallRepair_FencedJSON(t *testing.T) {
|
||||
t.Parallel()
|
||||
// A code-fenced tool-call envelope is repaired and executed rather than
|
||||
// silently becoming a text answer.
|
||||
call := "```json\n{\"tool_calls\":[{\"name\":\"echo\",\"args\":{\"q\":\"hi\"}}]}\n```"
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "done"}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
l := New(Options{Agent: a, MaxRounds: 8})
|
||||
|
||||
res, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")})
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
if res.ToolCallsExecuted != 1 || res.Text != "done" {
|
||||
t.Fatalf("ToolCallsExecuted = %d, Text = %q; want 1 + done (fenced JSON repaired)", res.ToolCallsExecuted, res.Text)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRun_StreamToolExecution_EmitsProgress(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
ad := &mockAdapter{
|
||||
stream: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: "done"}},
|
||||
},
|
||||
chunks: [][]string{{"x"}},
|
||||
}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
var mu sync.Mutex
|
||||
var evs []model.StreamEvent
|
||||
l := New(Options{
|
||||
Agent: a,
|
||||
OnStream: func(ev model.StreamEvent) {
|
||||
mu.Lock()
|
||||
evs = append(evs, ev)
|
||||
mu.Unlock()
|
||||
},
|
||||
MaxRounds: 8,
|
||||
})
|
||||
|
||||
if _, err := l.Run(context.Background(), "sess", []model.Message{model.TextMessage(model.RoleUser, "go")}); err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
saw := false
|
||||
for _, ev := range evs {
|
||||
if ev.Kind == model.StreamKindToolExecution {
|
||||
saw = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !saw {
|
||||
t.Fatalf("expected a tool_execution_update event on the streaming path, got %+v", evs)
|
||||
}
|
||||
}
|
||||
@@ -84,10 +84,13 @@ func TestRun_ToolCallThenFinal_FeedsBack(t *testing.T) {
|
||||
|
||||
func TestRun_MaxRounds(t *testing.T) {
|
||||
t.Parallel()
|
||||
call := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
// Distinct calls per round so the critical-tool-loop safety net does not
|
||||
// intervene; the goal here is to exercise the round budget.
|
||||
call1 := `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`
|
||||
call2 := `{"tool_calls":[{"name":"echo","args":{"q":"y"}}]}`
|
||||
ad := &mockAdapter{chat: []mgResp{
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: call}},
|
||||
{resp: model.Response{Content: call1}},
|
||||
{resp: model.Response{Content: call2}},
|
||||
}}
|
||||
td := dispatcherWith(t, &mockTool{name: "echo", result: "x"})
|
||||
a := newTestAgent(t, ad, td)
|
||||
|
||||
@@ -52,6 +52,16 @@ type Options struct {
|
||||
// forwarded to this callback (model.StreamEvent), so the frontend can render
|
||||
// body text, tool calls and (eventually) reasoning segments independently.
|
||||
OnStream func(model.StreamEvent)
|
||||
// ToolExecution selects how a batch of tool calls is run: "parallel" (the
|
||||
// default) or "sequential". Empty means parallel.
|
||||
ToolExecution string
|
||||
// ToolParallelism caps how many tool calls run concurrently when ToolExecution
|
||||
// is "parallel". Non-positive selects DefaultToolParallelism (4).
|
||||
ToolParallelism int
|
||||
// Hooks installs optional per-batch/per-call tool-loop hooks. Every field is
|
||||
// nil-tolerant; these are reserved for "敏感操作代审" (design/11) and the
|
||||
// future asynchronous-tool path. A zero Hooks value is a no-op.
|
||||
Hooks ToolLoopHooks
|
||||
// Log is the base logger. Nil is replaced with a discard logger tagged with
|
||||
// source "runtime/loop".
|
||||
Log *slog.Logger
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
// buildRequest assembles a model.Request from the agent's public surface,
|
||||
@@ -71,7 +70,11 @@ func decodeToolCalls(resp model.Response) ([]agent.ToolCall, bool) {
|
||||
var payload struct {
|
||||
ToolCalls []agent.ToolCall `json:"tool_calls"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(resp.Content), &payload); err != nil {
|
||||
// Lenient decode repairs common model-emitted malformations (code fences,
|
||||
// trailing commas) so a slightly broken tool-call envelope does not turn into
|
||||
// a text answer; an unrecoverable payload still yields ok=false and the round
|
||||
// proceeds as text.
|
||||
if err := tool.LenientJSONUnmarshal([]byte(resp.Content), &payload); err != nil {
|
||||
return nil, false
|
||||
}
|
||||
if len(payload.ToolCalls) == 0 {
|
||||
|
||||
@@ -30,6 +30,7 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
}
|
||||
|
||||
msgs := l.injectMemory(ctx, messages)
|
||||
toolState := &toolLoopState{prevFP: make(map[string]bool)}
|
||||
res := &Result{}
|
||||
for step := 1; step <= l.maxRounds; step++ {
|
||||
if err := ctx.Err(); err != nil {
|
||||
@@ -56,14 +57,24 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa
|
||||
return res, nil
|
||||
}
|
||||
|
||||
newMsgs, executed, err := l.runToolCalls(rc, msgs, outcome.toolCalls)
|
||||
batch, err := l.runToolCalls(rc, msgs, outcome.toolCalls, toolState)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("loop: run: step %d: tools: %w", step, err)
|
||||
}
|
||||
msgs = newMsgs
|
||||
res.ToolCallsExecuted += executed
|
||||
msgs = batch.messages
|
||||
res.ToolCallsExecuted += batch.executed
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, agent.StopReasonToolCalls)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
|
||||
// The critical-tool-loop safety net (or a terminating hook) stops the
|
||||
// round without waiting for a real final answer.
|
||||
if batch.terminate {
|
||||
res.Text = criticalToolLoopTerminalReason
|
||||
res.StopReason = StopReasonToolLoopTerminated
|
||||
l.saveCheckpoint(ctx, sessionID, step, msgs, StopReasonToolLoopTerminated)
|
||||
l.extractMemory(ctx, msgs, step)
|
||||
return res, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Budget exhausted without a final answer.
|
||||
|
||||
+395
-36
@@ -3,7 +3,9 @@ package loop
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
@@ -11,38 +13,378 @@ import (
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/trace"
|
||||
)
|
||||
|
||||
// runToolCalls executes each requested tool call, appending a RoleTool result
|
||||
// message (success or failure) so the model can continue. It returns the updated
|
||||
// message slice and the number of tool calls invoked. Invocation is delegated to
|
||||
// the dispatched tool, which applies the per-tool deadline; a missing dispatcher
|
||||
// is fed back as an error so the loop never panics or crashes on an unhandled
|
||||
// call.
|
||||
func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls []agent.ToolCall) ([]model.Message, int, error) {
|
||||
// ToolExecutionMode tokens accepted by Options.ToolExecution. Empty selects
|
||||
// toolExecParallel.
|
||||
const (
|
||||
toolExecParallel = "parallel"
|
||||
toolExecSequential = "sequential"
|
||||
)
|
||||
|
||||
// criticalToolLoopTerminalReason is surfaced as Result.Text and on every blocked
|
||||
// call when the critical-tool-loop safety net stops the run.
|
||||
const criticalToolLoopTerminalReason = "critical tool-loop detected: an identical tool call was repeated across rounds; the run is stopping"
|
||||
|
||||
// ToolCallVeto is returned by a CallAdmit hook to block a single call.
|
||||
type ToolCallVeto struct {
|
||||
Block bool
|
||||
Reason string
|
||||
}
|
||||
|
||||
// ToolOutcome is the normalized result of executing one tool call. It is what an
|
||||
// after-call / after-outcome hook observes and may rewrite before the loop
|
||||
// renders it back to the model.
|
||||
type ToolOutcome struct {
|
||||
// Result is the value the tool returned (nil on failure or skip).
|
||||
Result any
|
||||
// Err is non-nil when the call failed to execute; it is nil on success.
|
||||
Err error
|
||||
// ErrKind classifies a failure as tool.KindArgumentValidation or
|
||||
// tool.KindExecution (the task's errorKind status classification). Empty on
|
||||
// success.
|
||||
ErrKind string
|
||||
// Skipped is true when the call never executed (duplicate / blocked / vetoed).
|
||||
Skipped bool
|
||||
// Reason is a human-readable explanation for a skipped call.
|
||||
Reason string
|
||||
// Terminate, when set by a hook, signals that the whole round should stop
|
||||
// once every call in the batch is also marked terminate.
|
||||
Terminate bool
|
||||
}
|
||||
|
||||
// ToolLoopHooks groups the optional per-batch / per-call tool-loop hooks. Every
|
||||
// field is nil-tolerant. These are the extension points the loop keeps open for
|
||||
// "敏感操作代审" (design/11 batch admission) and the future async-tool path, so
|
||||
// the audit layer and the loop are decoupled.
|
||||
type ToolLoopHooks struct {
|
||||
// BatchAdmit is called once per batch before any call executes. Returning
|
||||
// (false, reason) blocks the whole batch: every call becomes a skipped result
|
||||
// carrying reason. nil = always admit.
|
||||
BatchAdmit func(ctx context.Context, calls []agent.ToolCall) (bool, string)
|
||||
// CallAdmit is called per call before it executes. A non-nil veto with
|
||||
// Block=true blocks that single call. nil = always admit.
|
||||
CallAdmit func(ctx context.Context, call agent.ToolCall) *ToolCallVeto
|
||||
// AfterCall rewrites a call's outcome after the tool executes. A nil return
|
||||
// keeps the original. It runs before AfterOutcome.
|
||||
AfterCall func(ctx context.Context, call agent.ToolCall, outcome ToolOutcome) *ToolOutcome
|
||||
// AfterOutcome observes/rewrites the finalized outcome (after AfterCall and
|
||||
// after any skip/block decision) immediately before the result is rendered
|
||||
// back to the model. nil = keep the outcome.
|
||||
AfterOutcome func(ctx context.Context, call agent.ToolCall, outcome ToolOutcome) *ToolOutcome
|
||||
}
|
||||
|
||||
// toolRunResult is the outcome of running one tool batch.
|
||||
type toolRunResult struct {
|
||||
messages []model.Message
|
||||
executed int
|
||||
terminate bool
|
||||
}
|
||||
|
||||
// toolLoopState carries the per-run state of the critical-tool-loop safety net.
|
||||
// It is scoped to a single Run call, so concurrent runs never share it.
|
||||
type toolLoopState struct {
|
||||
criticalSeen bool // a critical loop was already reported this run
|
||||
prevFP map[string]bool // fingerprints seen in the immediately previous batch
|
||||
}
|
||||
|
||||
// toolCallAction is the execution decision for one call after classification.
|
||||
type toolCallAction struct {
|
||||
run bool // execute the call
|
||||
reason string // when run==false, the human reason fed back to the model
|
||||
}
|
||||
|
||||
// runToolCalls executes a batch of model-requested tool calls. It classifies
|
||||
// duplicate invocations (the critical-tool-loop safety net), applies the optional
|
||||
// batch/per-call hooks, runs the surviving calls in parallel (bounded) or
|
||||
// serially, and feeds every outcome back as a RoleTool message in the original
|
||||
// call order. It returns the number of calls actually invoked (executed), the
|
||||
// updated message slice, and whether the round should terminate.
|
||||
//
|
||||
// Invocation is delegated to the dispatched tool, which applies the per-tool
|
||||
// deadline; a missing dispatcher is fed back as an error so the loop never
|
||||
// panics or crashes on an unhandled call.
|
||||
func (l *Loop) runToolCalls(ctx context.Context, messages []model.Message, calls []agent.ToolCall, st *toolLoopState) (toolRunResult, error) {
|
||||
d := l.dispatcher()
|
||||
out := messages
|
||||
executed := 0
|
||||
for _, tc := range calls {
|
||||
var (
|
||||
result any
|
||||
invokeErr error
|
||||
)
|
||||
if d == nil {
|
||||
invokeErr = fmt.Errorf("no tool dispatcher configured for %q", tc.Name)
|
||||
} else {
|
||||
_, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool)
|
||||
result, invokeErr = d.Invoke(ctx, tc.Name, tc.Args)
|
||||
if invokeErr != nil {
|
||||
l.spanProp(toolSpan, "error", invokeErr.Error())
|
||||
l.finishSpan(toolSpan, trace.SpanStatusError)
|
||||
} else {
|
||||
l.spanProp(toolSpan, "result", result)
|
||||
l.finishSpan(toolSpan, trace.SpanStatusOK)
|
||||
res := toolRunResult{messages: messages}
|
||||
if len(calls) == 0 {
|
||||
return res, nil
|
||||
}
|
||||
|
||||
actions, terminate := classifyToolBatch(st, calls)
|
||||
if terminate {
|
||||
res.terminate = true
|
||||
}
|
||||
|
||||
// BatchAdmit hook (audit / political review): block the whole batch.
|
||||
if l.hooks.BatchAdmit != nil {
|
||||
if ok, reason := l.hooks.BatchAdmit(ctx, calls); !ok {
|
||||
if reason == "" {
|
||||
reason = "tool batch blocked by policy"
|
||||
}
|
||||
for i := range calls {
|
||||
actions[i] = toolCallAction{run: false, reason: reason}
|
||||
}
|
||||
}
|
||||
executed++
|
||||
out = append(out, toolResultMessage(tc, result, invokeErr))
|
||||
}
|
||||
return out, executed, nil
|
||||
|
||||
// CallAdmit hook: per-call veto over the calls we intend to run.
|
||||
if l.hooks.CallAdmit != nil {
|
||||
for i := range calls {
|
||||
if !actions[i].run {
|
||||
continue
|
||||
}
|
||||
if veto := l.hooks.CallAdmit(ctx, calls[i]); veto != nil && veto.Block {
|
||||
reason := veto.Reason
|
||||
if reason == "" {
|
||||
reason = "tool call blocked by policy"
|
||||
}
|
||||
actions[i] = toolCallAction{run: false, reason: reason}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Decide serial vs parallel: explicit sequential mode, or any call whose tool
|
||||
// is declared sequential (a single sequential marker makes the whole group
|
||||
// serial, matching OpenClaw's launchParallelToolCalls).
|
||||
sequential := l.toolExec == toolExecSequential
|
||||
if !sequential && d != nil {
|
||||
for i := range calls {
|
||||
if actions[i].run && d.IsSequential(calls[i].Name) {
|
||||
sequential = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
outcomes := make([]ToolOutcome, len(calls))
|
||||
if sequential {
|
||||
for i := range calls {
|
||||
if actions[i].run {
|
||||
outcomes[i] = l.invokeOne(ctx, d, calls[i])
|
||||
res.executed++
|
||||
} else {
|
||||
outcomes[i] = skippedOutcome(actions[i].reason)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
idxs := make([]int, 0, len(calls))
|
||||
for i := range calls {
|
||||
if actions[i].run {
|
||||
idxs = append(idxs, i)
|
||||
} else {
|
||||
outcomes[i] = skippedOutcome(actions[i].reason)
|
||||
}
|
||||
}
|
||||
l.invokeParallel(ctx, d, calls, idxs, outcomes)
|
||||
res.executed += len(idxs)
|
||||
}
|
||||
|
||||
// AfterCall hook (only for calls that actually executed).
|
||||
if l.hooks.AfterCall != nil {
|
||||
for i := range calls {
|
||||
if outcomes[i].Skipped {
|
||||
continue
|
||||
}
|
||||
if after := l.hooks.AfterCall(ctx, calls[i], outcomes[i]); after != nil {
|
||||
outcomes[i] = *after
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// AfterOutcome hook (runs on every finalized outcome, skipped or not).
|
||||
if l.hooks.AfterOutcome != nil {
|
||||
for i := range calls {
|
||||
if after := l.hooks.AfterOutcome(ctx, calls[i], outcomes[i]); after != nil {
|
||||
outcomes[i] = *after
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A hook may ask to terminate the round by marking every call terminate
|
||||
// (OpenClaw's shouldTerminateToolBatch).
|
||||
if !res.terminate && shouldTerminateBatch(outcomes) {
|
||||
res.terminate = true
|
||||
}
|
||||
|
||||
// Render results in the original call order so the model sees a stable feed.
|
||||
for i := range calls {
|
||||
res.messages = append(res.messages, toolResultMessage(calls[i], outcomes[i]))
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
|
||||
// classifyToolBatch partitions a batch into run/skip decisions, enforcing the
|
||||
// critical-tool-loop safety net:
|
||||
//
|
||||
// - A call identical (same name + arguments) to one earlier in the same batch
|
||||
// is skipped with a duplicate hint (never executes).
|
||||
// - A call identical to one from a previous round is a critical-loop repeat:
|
||||
// the first occurrence blocks the repeated call and sets criticalSeen, so the
|
||||
// model gets a turn to reassess; a second consecutive occurrence terminates
|
||||
// the whole run.
|
||||
//
|
||||
// It returns (actions, terminate). When terminate is true every call is blocked
|
||||
// with criticalToolLoopTerminalReason and the round must end.
|
||||
func classifyToolBatch(st *toolLoopState, calls []agent.ToolCall) ([]toolCallAction, bool) {
|
||||
seen := make(map[string]bool, len(calls))
|
||||
actions := make([]toolCallAction, len(calls))
|
||||
repeat := false
|
||||
for i, call := range calls {
|
||||
fp := toolFingerprint(call)
|
||||
if seen[fp] {
|
||||
// Duplicate of a call already in this batch.
|
||||
if st.prevFP[fp] {
|
||||
actions[i] = toolCallAction{run: false, reason: repeatedCallReason()}
|
||||
repeat = true
|
||||
} else {
|
||||
actions[i] = toolCallAction{run: false, reason: duplicateCallReason()}
|
||||
}
|
||||
continue
|
||||
}
|
||||
seen[fp] = true
|
||||
if st.prevFP[fp] {
|
||||
// Repeated from the previous round — a critical loop.
|
||||
actions[i] = toolCallAction{run: false, reason: repeatedCallReason()}
|
||||
repeat = true
|
||||
} else {
|
||||
actions[i] = toolCallAction{run: true}
|
||||
}
|
||||
}
|
||||
|
||||
// Remember which fingerprints were live this round for the next round's
|
||||
// cross-round repeat detection.
|
||||
st.prevFP = seen
|
||||
|
||||
terminate := false
|
||||
if repeat {
|
||||
if st.criticalSeen {
|
||||
terminate = true
|
||||
for i := range actions {
|
||||
actions[i] = toolCallAction{run: false, reason: criticalToolLoopTerminalReason}
|
||||
}
|
||||
} else {
|
||||
st.criticalSeen = true
|
||||
}
|
||||
}
|
||||
return actions, terminate
|
||||
}
|
||||
|
||||
// invokeOne executes a single tool call under its per-tool deadline, emitting a
|
||||
// tool span, streaming tool_execution_update progress and normalizing the
|
||||
// outcome.
|
||||
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}
|
||||
}
|
||||
l.emitToolExecution(tc.Name, "start")
|
||||
_, toolSpan := l.span(ctx, "tool."+tc.Name, trace.SpanKindTool)
|
||||
result, err := d.Invoke(ctx, tc.Name, tc.Args)
|
||||
phase := "end"
|
||||
if err != nil {
|
||||
phase = "end:error"
|
||||
}
|
||||
l.emitToolExecution(tc.Name, phase)
|
||||
out := ToolOutcome{Result: result, Err: err, ErrKind: errorKind(err)}
|
||||
if err != nil {
|
||||
l.spanProp(toolSpan, "error", err.Error())
|
||||
l.spanProp(toolSpan, "error_kind", out.ErrKind)
|
||||
l.finishSpan(toolSpan, trace.SpanStatusError)
|
||||
} else {
|
||||
l.spanProp(toolSpan, "result", result)
|
||||
l.finishSpan(toolSpan, trace.SpanStatusOK)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// emitToolExecution forwards a tool_execution_update progress event to the
|
||||
// frontend when the loop is on the streaming path (OnStream configured); it is a
|
||||
// no-op otherwise. phase is a short label like "start" or "end:error".
|
||||
func (l *Loop) emitToolExecution(name, phase string) {
|
||||
if l.onStream == nil {
|
||||
return
|
||||
}
|
||||
l.onStream(model.StreamEvent{Kind: model.StreamKindToolExecution, Delta: phase + ":" + name})
|
||||
}
|
||||
|
||||
// invokeParallel runs calls at idxs concurrently, capped at the loop's
|
||||
// parallelism, writing each outcome back into out at its call index so callers
|
||||
// can render results in the original order. The semaphore bounds concurrency and
|
||||
// in-flight goroutines release their slot on completion, so a canceled context
|
||||
// (which makes every Invoke return promptly) cannot leak goroutines.
|
||||
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()
|
||||
}
|
||||
|
||||
// returned by classifyToolBatch for the two skip reasons.
|
||||
func duplicateCallReason() string {
|
||||
return "duplicate call with identical arguments already issued in this batch"
|
||||
}
|
||||
|
||||
func repeatedCallReason() string {
|
||||
return "already used with identical arguments in a previous round; do not repeat it"
|
||||
}
|
||||
|
||||
// skippedOutcome builds a Skipped outcome carrying the human reason.
|
||||
func skippedOutcome(reason string) ToolOutcome {
|
||||
if reason == "" {
|
||||
reason = "tool call skipped"
|
||||
}
|
||||
return ToolOutcome{Skipped: true, Reason: reason}
|
||||
}
|
||||
|
||||
// shouldTerminateBatch reports whether every call in the batch asked to terminate
|
||||
// (OpenClaw's shouldTerminateToolBatch), so a hook can stop the round.
|
||||
func shouldTerminateBatch(outcomes []ToolOutcome) bool {
|
||||
if len(outcomes) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, o := range outcomes {
|
||||
if !o.Terminate {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// toolFingerprint canonicalizes a tool call (name + args) so identical calls can
|
||||
// be detected. Args are JSON-marshaled, so equivalent objects with different key
|
||||
// orders collapse to the same fingerprint; a marshal failure falls back to %v.
|
||||
func toolFingerprint(call agent.ToolCall) string {
|
||||
b, err := json.Marshal(call.Args)
|
||||
if err != nil {
|
||||
b = []byte(fmt.Sprintf("%v", call.Args))
|
||||
}
|
||||
return call.Name + "\x00" + string(b)
|
||||
}
|
||||
|
||||
// errorKind classifies an invocation error into a tool.Kind* label so the model
|
||||
// can distinguish an argument-validation failure (re-issue a corrected call) from
|
||||
// an execution failure. A nil error yields "" (success).
|
||||
func errorKind(err error) string {
|
||||
if err == nil {
|
||||
return ""
|
||||
}
|
||||
var te *tool.ToolError
|
||||
if errors.As(err, &te) {
|
||||
return te.Kind()
|
||||
}
|
||||
if errors.Is(err, tool.ErrInvalidArgs) {
|
||||
return tool.KindArgumentValidation
|
||||
}
|
||||
return tool.KindExecution
|
||||
}
|
||||
|
||||
// dispatcher returns the loop's tool dispatcher, falling back to the agent's own
|
||||
@@ -58,15 +400,32 @@ func (l *Loop) dispatcher() *tool.Dispatcher {
|
||||
}
|
||||
|
||||
// toolResultMessage renders a tool call's outcome as a RoleTool message ready to
|
||||
// be fed back to the model. Errors are included in the text so the LLM can
|
||||
// react to a failed or timed-out call.
|
||||
func toolResultMessage(tc agent.ToolCall, result any, invokeErr error) model.Message {
|
||||
text := fmt.Sprintf("tool %s result: %s", tc.Name, formatToolResult(result))
|
||||
if invokeErr != nil {
|
||||
text = fmt.Sprintf("tool %s failed: %v", tc.Name, invokeErr)
|
||||
}
|
||||
// be fed back to the model. Skipped calls carry their reason; failures carry a
|
||||
// structured {kind,message} so the LLM can react appropriately; success carries
|
||||
// the value.
|
||||
func toolResultMessage(tc agent.ToolCall, out ToolOutcome) model.Message {
|
||||
return model.Message{Role: model.RoleTool, ToolCallName: tc.Name,
|
||||
Content: []model.Content{{Type: model.ContentTypeText, Text: text}}}
|
||||
Content: []model.Content{{Type: model.ContentTypeText, Text: toolResultText(tc, out)}}}
|
||||
}
|
||||
|
||||
// toolResultText renders the text payload of a tool result message.
|
||||
func toolResultText(tc agent.ToolCall, out ToolOutcome) string {
|
||||
switch {
|
||||
case out.Skipped:
|
||||
if out.Reason == "" {
|
||||
return fmt.Sprintf("tool %s skipped", tc.Name)
|
||||
}
|
||||
return fmt.Sprintf("tool %s skipped: %s", tc.Name, out.Reason)
|
||||
case out.Err != nil:
|
||||
if kind := out.ErrKind; kind != "" {
|
||||
if payload, err := json.Marshal(map[string]string{"kind": kind, "message": out.Err.Error()}); err == nil {
|
||||
return fmt.Sprintf("tool %s failed: %s", tc.Name, payload)
|
||||
}
|
||||
}
|
||||
return fmt.Sprintf("tool %s failed: %v", tc.Name, out.Err)
|
||||
default:
|
||||
return fmt.Sprintf("tool %s result: %s", tc.Name, formatToolResult(out.Result))
|
||||
}
|
||||
}
|
||||
|
||||
// formatToolResult renders a tool return value as text for the model. Strings are
|
||||
|
||||
@@ -0,0 +1,113 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"regexp"
|
||||
)
|
||||
|
||||
// fencedJSONRe matches a code-fence wrapper some models emit around tool-call
|
||||
// JSON, e.g. ```json {...} ```. The captured group is the inner JSON.
|
||||
var fencedJSONRe = regexp.MustCompile("(?s)^[\\s`]*```(?:json)?[\\r\\n]*(.*?)[\\r\\n]*```\\s*$")
|
||||
|
||||
// LenientJSONUnmarshal unmarshals raw into dest, applying a few tolerant repairs
|
||||
// for the malformed tool-call JSON models sometimes emit, before falling back to
|
||||
// strict encoding/json. Repairs are applied in order:
|
||||
//
|
||||
// 1. strip a wrapping ```json code fence (and stray backticks);
|
||||
// 2. trim surrounding whitespace;
|
||||
// 3. strip trailing commas before `}` and `]`;
|
||||
// 4. attempt the strict decode.
|
||||
//
|
||||
// The function reports the raw json.Unmarshal error (wrapped) when every repair
|
||||
// fails, so a caller can fall back to treating the payload as plain text without
|
||||
// crashing — the malformed call never aborts the round. Full repair (unquoted
|
||||
// keys, single-quoted strings, smart quotes) is deliberately left for a later
|
||||
// milestone; the load-bearing guarantee here is that a malformed envelope is
|
||||
// recoverable or degrades to text rather than throwing.
|
||||
func LenientJSONUnmarshal(raw []byte, dest any) error {
|
||||
candidates := repairCandidates(raw)
|
||||
var lastErr error
|
||||
for _, c := range candidates {
|
||||
if err := json.Unmarshal(c, dest); err == nil {
|
||||
return nil
|
||||
} else {
|
||||
lastErr = err
|
||||
}
|
||||
}
|
||||
return fmt.Errorf("tool: lenient json: %w", lastErr)
|
||||
}
|
||||
|
||||
// repairCandidates returns the ordered list of byte slices to try, deduplicated:
|
||||
// the raw input, the fenced-stripped input, and trailing-comma-stripped variants.
|
||||
func repairCandidates(raw []byte) [][]byte {
|
||||
out := [][]byte{nil}
|
||||
seen := map[string]bool{}
|
||||
push := func(b []byte) {
|
||||
s := string(b)
|
||||
if seen[s] {
|
||||
return
|
||||
}
|
||||
seen[s] = true
|
||||
out = append(out, b)
|
||||
}
|
||||
|
||||
push(bytes.TrimSpace(raw))
|
||||
|
||||
clean := raw
|
||||
if m := fencedJSONRe.FindSubmatch(raw); len(m) == 2 {
|
||||
clean = bytes.TrimSpace(m[1])
|
||||
push(clean)
|
||||
}
|
||||
clean = bytes.TrimSpace(clean)
|
||||
push(bytes.TrimSpace(clean))
|
||||
push(bytes.TrimSpace(stripTrailingCommas(clean)))
|
||||
push(bytes.TrimSpace(stripTrailingCommas(bytes.TrimSpace(raw))))
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// stripTrailingCommas removes commas that sit immediately before a closing `}`
|
||||
// or `]` (ignoring whitespace), the most common single-token repair for truncated
|
||||
// model output such as `{"a":1,}`.
|
||||
func stripTrailingCommas(b []byte) []byte {
|
||||
if !bytes.ContainsAny(b, "}]") {
|
||||
return b
|
||||
}
|
||||
// Scan for a comma run whose next non-whitespace byte is a closing bracket; a
|
||||
// copy is built so the original bytes are never mutated.
|
||||
out := make([]byte, 0, len(b))
|
||||
skipComma := false
|
||||
for i := 0; i < len(b); i++ {
|
||||
if b[i] == ',' {
|
||||
// Peek ahead past whitespace for a closing bracket.
|
||||
j := i + 1
|
||||
for j < len(b) && isJSONSpace(b[j]) {
|
||||
j++
|
||||
}
|
||||
if j < len(b) && (b[j] == '}' || b[j] == ']') {
|
||||
skipComma = true
|
||||
}
|
||||
}
|
||||
if skipComma && isJSONSpace(b[i]) {
|
||||
continue
|
||||
}
|
||||
if skipComma && b[i] == ',' {
|
||||
skipComma = false
|
||||
continue
|
||||
}
|
||||
out = append(out, b[i])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// isJSONSpace reports whether b is JSON insignificant whitespace.
|
||||
func isJSONSpace(b byte) bool {
|
||||
switch b {
|
||||
case ' ', '\t', '\r', '\n':
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
@@ -25,6 +25,28 @@ import (
|
||||
// override and the Dispatcher was built without WithDefaultTimeout.
|
||||
const DefaultTimeout = 30 * time.Second
|
||||
|
||||
// Execution modes a tool may declare to the loop. A tool that mutates shared
|
||||
// state (file writes, device control) should be marked sequential so the loop
|
||||
// does not run it concurrently with other calls.
|
||||
const (
|
||||
// ExecutionModeParallel is the default: the loop may run such a tool
|
||||
// concurrently with other calls in a batch.
|
||||
ExecutionModeParallel = "parallel"
|
||||
// ExecutionModeSequential forces the loop to run the whole batch serially so
|
||||
// this tool is never invoked concurrently.
|
||||
ExecutionModeSequential = "sequential"
|
||||
)
|
||||
|
||||
// Error-kind labels distinguishing why a tool call failed, fed back to the model
|
||||
// as a structured {kind,message} so it can react differently to a bad argument
|
||||
// (re-issue a corrected call) versus an execution failure.
|
||||
const (
|
||||
// KindArgumentValidation marks a JSON-Schema argument-validation failure.
|
||||
KindArgumentValidation = "argument-validation"
|
||||
// KindExecution marks a runtime execution failure (timeout, panic, tool error).
|
||||
KindExecution = "execution"
|
||||
)
|
||||
|
||||
// Sentinel errors returned by the Dispatcher. Use errors.Is to test for them;
|
||||
// they are wrapped with %w so the underlying cause stays available.
|
||||
var (
|
||||
@@ -64,6 +86,17 @@ func (e *ToolError) Error() string {
|
||||
// Unwrap exposes the wrapped cause so errors.Is/errors.As traverse the chain.
|
||||
func (e *ToolError) Unwrap() error { return e.cause }
|
||||
|
||||
// Kind classifies the failure as "argument-validation" (the caller-supplied
|
||||
// arguments failed the JSON Schema) or "execution" (anything after validation:
|
||||
// unknown tool, timeout, panic or an Execute error). The loop uses it to feed a
|
||||
// structured {kind,message} tool result back to the model.
|
||||
func (e *ToolError) Kind() string {
|
||||
if e.cause != nil && errors.Is(e.cause, ErrInvalidArgs) {
|
||||
return KindArgumentValidation
|
||||
}
|
||||
return KindExecution
|
||||
}
|
||||
|
||||
// Option configures a Dispatcher at construction time.
|
||||
type Option func(*Dispatcher)
|
||||
|
||||
@@ -72,8 +105,9 @@ type Option func(*Dispatcher)
|
||||
type Dispatcher struct {
|
||||
reg *registry.Registry
|
||||
dft time.Duration
|
||||
mu sync.RWMutex // guards timeouts
|
||||
mu sync.RWMutex // guards timeouts and modes
|
||||
timeouts map[string]time.Duration
|
||||
modes map[string]string // tool name -> ExecutionMode* (default parallel)
|
||||
}
|
||||
|
||||
// New builds a Dispatcher backed by reg, applying opts in order. It returns an
|
||||
@@ -87,6 +121,7 @@ func New(reg *registry.Registry, opts ...Option) (*Dispatcher, error) {
|
||||
reg: reg,
|
||||
dft: DefaultTimeout,
|
||||
timeouts: make(map[string]time.Duration),
|
||||
modes: make(map[string]string),
|
||||
}
|
||||
for _, opt := range opts {
|
||||
if opt != nil {
|
||||
@@ -125,6 +160,33 @@ func (d *Dispatcher) SetToolTimeout(name string, timeout time.Duration) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetToolExecutionMode marks a tool as running "sequential" or "parallel" (the
|
||||
// default). A sequential tool is never invoked concurrently by the loop, so
|
||||
// shared-state tools (file writes, device control) can declare it. An empty mode
|
||||
// clears the override back to the default. Only ExecutionModeSequential is
|
||||
// treated as sequential.
|
||||
func (d *Dispatcher) SetToolExecutionMode(name, mode string) error {
|
||||
if name == "" {
|
||||
return fmt.Errorf("tool: set execution mode: %w", ErrUnknownTool)
|
||||
}
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if mode == "" || mode == ExecutionModeParallel {
|
||||
delete(d.modes, name)
|
||||
return nil
|
||||
}
|
||||
d.modes[name] = mode
|
||||
return nil
|
||||
}
|
||||
|
||||
// IsSequential reports whether the named tool has been declared sequential. An
|
||||
// unknown name or one never marked sequential reports false (the default).
|
||||
func (d *Dispatcher) IsSequential(name string) bool {
|
||||
d.mu.RLock()
|
||||
defer d.mu.RUnlock()
|
||||
return d.modes[name] == ExecutionModeSequential
|
||||
}
|
||||
|
||||
// Invoke executes the tool registered as name with the given args. It resolves
|
||||
// the tool from the registry, validates args against the tool's Parameters JSON
|
||||
// Schema, and runs Execute under the tool's deadline (per-tool override, else
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package tool
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestLenientJSONUnmarshal_TrailingComma(t *testing.T) {
|
||||
t.Parallel()
|
||||
var v map[string]any
|
||||
if err := LenientJSONUnmarshal([]byte(`{"a":1,}`), &v); err != nil {
|
||||
t.Fatalf("LenientJSONUnmarshal: %v", err)
|
||||
}
|
||||
if v["a"] != float64(1) {
|
||||
t.Fatalf("a = %#v, want 1", v["a"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestLenientJSONUnmarshal_Fenced(t *testing.T) {
|
||||
t.Parallel()
|
||||
raw := "```json\n{\"tool_calls\":[{\"name\":\"echo\",\"args\":{\"q\":\"hi\"}}]}\n```"
|
||||
var v struct {
|
||||
ToolCalls []map[string]any `json:"tool_calls"`
|
||||
}
|
||||
if err := LenientJSONUnmarshal([]byte(raw), &v); err != nil {
|
||||
t.Fatalf("LenientJSONUnmarshal: %v", err)
|
||||
}
|
||||
if len(v.ToolCalls) != 1 {
|
||||
t.Fatalf("tool_calls = %d, want 1", len(v.ToolCalls))
|
||||
}
|
||||
}
|
||||
|
||||
func TestLenientJSONUnmarshal_Unrecoverable(t *testing.T) {
|
||||
t.Parallel()
|
||||
// Malformed beyond repair must return an error, never panic or silently
|
||||
// succeed.
|
||||
var v map[string]any
|
||||
if err := LenientJSONUnmarshal([]byte(`{"a":"unterminated`), &v); err == nil {
|
||||
t.Fatal("expected error for unrecoverable JSON")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetToolExecutionMode_Sequential(t *testing.T) {
|
||||
t.Parallel()
|
||||
d, _ := newDispatcher(t)
|
||||
if d.IsSequential("x") {
|
||||
t.Fatal("x should default to parallel")
|
||||
}
|
||||
if err := d.SetToolExecutionMode("x", ExecutionModeSequential); err != nil {
|
||||
t.Fatalf("SetToolExecutionMode: %v", err)
|
||||
}
|
||||
if !d.IsSequential("x") {
|
||||
t.Fatal("x should be sequential after marking")
|
||||
}
|
||||
// Clearing back to parallel.
|
||||
if err := d.SetToolExecutionMode("x", ExecutionModeParallel); err != nil {
|
||||
t.Fatalf("SetToolExecutionMode(parallel): %v", err)
|
||||
}
|
||||
if d.IsSequential("x") {
|
||||
t.Fatal("x should be parallel after clearing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetToolExecutionMode_EmptyName(t *testing.T) {
|
||||
t.Parallel()
|
||||
d, _ := newDispatcher(t)
|
||||
if err := d.SetToolExecutionMode("", ExecutionModeSequential); !errors.Is(err, ErrUnknownTool) {
|
||||
t.Fatalf("SetToolExecutionMode(\"\") error = %v, want ErrUnknownTool", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestToolError_Kind(t *testing.T) {
|
||||
t.Parallel()
|
||||
val := &ToolError{Name: "x", Message: "bad args", cause: fmt.Errorf("validate: %w", ErrInvalidArgs)}
|
||||
if val.Kind() != KindArgumentValidation {
|
||||
t.Fatalf("validation Kind = %q, want %q", val.Kind(), KindArgumentValidation)
|
||||
}
|
||||
exec := &ToolError{Name: "x", Message: "boom", cause: errors.New("boom")}
|
||||
if exec.Kind() != KindExecution {
|
||||
t.Fatalf("execution Kind = %q, want %q", exec.Kind(), KindExecution)
|
||||
}
|
||||
none := &ToolError{Name: "x", Message: "m"}
|
||||
if none.Kind() != KindExecution {
|
||||
t.Fatalf("nil-cause Kind = %q, want %q", none.Kind(), KindExecution)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user