diff --git a/internal/runtime/hostbridge.go b/internal/runtime/hostbridge.go index 2b4e40a..21d8063 100644 --- a/internal/runtime/hostbridge.go +++ b/internal/runtime/hostbridge.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "sync" "agent-native/adapters/native" "agent-native/runtime/tools" @@ -25,14 +26,30 @@ type producedSink interface { add(ref *core.ArtifactRef) } +// artifactList 是本次运行已登记的工件表。宿主工具可能被内核并行调用， +// 因此写入与读取都在同一把锁下：并发调用既不丢条目也不读到半成品。 type artifactList struct { + mu sync.Mutex items []*core.ArtifactRef } func (a *artifactList) add(ref *core.ArtifactRef) { - if ref != nil { - a.items = append(a.items, ref) + if ref == nil { + return } + a.mu.Lock() + defer a.mu.Unlock() + a.items = append(a.items, ref) +} + +// snapshot 返回当前工件的副本，供终局检查与部件装配使用。 +func (a *artifactList) snapshot() []*core.ArtifactRef { + if a == nil { + return nil + } + a.mu.Lock() + defer a.mu.Unlock() + return append([]*core.ArtifactRef(nil), a.items...) } // bridgeHostTools 把画像允许的产品工具包成内核宿主工具。 diff --git a/internal/runtime/squad_events.go b/internal/runtime/squad_events.go new file mode 100644 index 0000000..15cdfa7 --- /dev/null +++ b/internal/runtime/squad_events.go @@ -0,0 +1,188 @@ +package runtime + +import ( + "strings" + "sync" + + "agent-squad/squad" + core "agent-system-core" +) + +// toolResultSummaryLen 是工具结果摘要的截断长度（与既有展示一致）。 +const toolResultSummaryLen = 160 + +// toolEventID 是宿主侧的工具调用身份：优先用上游 CallID（并行同名工具据此配对， +// 而不是按到达顺序）；空串 = legacy/未知，退回工具名，保持既有展示不变。 +func toolEventID(ev squad.ProcessEvent) string { + if id := strings.TrimSpace(ev.CallID); id != "" { + return id + } + return ev.Tool +} + +// toolEventOutcome 把上游 tool_execution_end 翻译成宿主展示的真实结局。 +// +// 取值域与上游一致（ok | error | denied | canceled），不新增、不改写上游状态： +// - Status 非空：原样透传； +// - Status 为空且没有失败信号（ErrorReason 也空）：legacy/未知，沿用旧的 "ok" +// 与 Detail 摘要（字节一致，不臆造结局）； +// - Status 为空但有 ErrorReason：有失败信号而状态未知，报 error 而不是 ok； +// - 非 ok 时摘要取 ErrorReason 原文（截断），上游没给原因才回退 Detail。 +func toolEventOutcome(ev squad.ProcessEvent) (status, summary string) { + status = strings.TrimSpace(ev.Status) + if status == "" { + if strings.TrimSpace(ev.ErrorReason) == "" { + return squad.ToolStatusOK, truncateRun(ev.Detail, toolResultSummaryLen) + } + status = squad.ToolStatusError + } + if status == squad.ToolStatusOK { + return status, truncateRun(ev.Detail, toolResultSummaryLen) + } + if strings.TrimSpace(ev.ErrorReason) != "" { + return status, truncateRun(ev.ErrorReason, toolResultSummaryLen) + } + return status, truncateRun(ev.Detail, toolResultSummaryLen) +} + +// toolFailure 是一次未被恢复的工具失败（用于终局文本一致性检查，不覆盖工具部件）。 +type toolFailure struct { + Tool string + CallID string + Status string + Summary string +} + +// squadEventLog 是小队过程帧在宿主侧的聚合点。 +// +// 并行工具调用会从多个 goroutine 回调进来（内核 executeParallel 逐个派发）， +// 因此流式文本、工具部件、用量与 pending 配对表全部在同一把锁下更新： +// 既不丢条目，也不会把 A 调用的 start 与 B 调用的 end 交错写坏。 +// 事件发射与部件追加在同一临界区内，保持到达顺序与最终部件顺序一致。 +type squadEventLog struct { + rc RunCtx + + mu sync.Mutex + streamed strings.Builder + textIdx int + parts []core.Part + pending map[string]string // callID → 工具名（start/end 配对） + failure *toolFailure // 最近一次失败；同名工具后续再起即视为已恢复 +} + +// newSquadEventLog 绑定一次运行的过程帧回调。 +func newSquadEventLog(rc RunCtx) *squadEventLog { + return &squadEventLog{rc: rc, pending: map[string]string{}} +} + +// onEvent 是小队过程帧回调（并发安全）。 +func (l *squadEventLog) onEvent(ev squad.ProcessEvent) { + l.mu.Lock() + defer l.mu.Unlock() + switch ev.Type { + case "text_delta": + if ev.Text == "" { + return + } + l.streamed.WriteString(ev.Text) + _ = l.rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: l.textIdx, Text: ev.Text}) + case "reasoning_delta": + if ev.Text != "" { + _ = l.rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Text: ev.Text}) + } + case "tool_execution_start": + id := toolEventID(ev) + if id != "" { + l.pending[id] = ev.Tool + } + l.noteRecovered(ev.Tool) + _ = l.rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: id, Name: ev.Tool, Arguments: ev.Detail}) + l.parts = append(l.parts, core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: id, Name: ev.Tool, Arguments: ev.Detail}}) + case "tool_execution_end": + id := toolEventID(ev) + if tool, ok := l.pending[id]; ok && ev.Tool == "" { + ev.Tool = tool + } + delete(l.pending, id) + status, summary := toolEventOutcome(ev) + // 发射与部件同源：前端过程行与落库部件说的是同一件事。 + _ = l.rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: id, Status: status, Summary: summary}) + l.parts = append(l.parts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: id, Status: status, Summary: summary}}) + if status == squad.ToolStatusOK { + l.noteRecovered(ev.Tool) + } else { + l.failure = &toolFailure{Tool: ev.Tool, CallID: id, Status: status, Summary: summary} + } + case "usage": + if sink, ok := l.rc.(interface{ AddUsage(core.Usage) }); ok { + sink.AddUsage(core.Usage{ + PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens, + LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens, + CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens, + }) + } + } +} + +// noteRecovered 把同名工具的后续进展（再次发起或调用成功）记为已恢复： +// 失败部件仍留在过程部件里，只是不再让收尾文本替它宣布成功。 +func (l *squadEventLog) noteRecovered(tool string) { + if l.failure != nil && l.failure.Tool == tool { + l.failure = nil + } +} + +// reply 是本次运行聚合到的模型文本（不含任何宿主补写）。 +func (l *squadEventLog) reply() string { + l.mu.Lock() + defer l.mu.Unlock() + return l.streamed.String() +} + +// parts 返回过程部件的快照（复制，避免把内部切片交给最终部件）。 +func (l *squadEventLog) toolParts() []core.Part { + l.mu.Lock() + defer l.mu.Unlock() + return append([]core.Part(nil), l.parts...) +} + +// unresolvedFailure 返回终局时仍未恢复的工具失败，没有则 nil。 +func (l *squadEventLog) unresolvedFailure() *toolFailure { + l.mu.Lock() + defer l.mu.Unlock() + if l.failure == nil { + return nil + } + f := *l.failure + return &f +} + +// finalReply 决定终局文本。模型文本原样保留（不代写、不改写）；只有宿主兜底的 +// 「已完成。」在仍有未恢复的工具失败时会被换成如实说明，避免收尾文本与失败部件 +// 自相矛盾。legacy 事件（无 CallID/Status）走不到这里，展示与旧行为一致。 +func (l *squadEventLog) finalReply(modelText, summary, fallback string) string { + reply := strings.TrimSpace(modelText) + if reply == "" { + reply = strings.TrimSpace(summary) + } + if reply != "" { + return reply + } + if f := l.unresolvedFailure(); f != nil { + return failedToolLine(f) + } + return fallback +} + +// failedToolLine 如实描述未恢复的工具失败，不声称成功，也不编造原因。 +func failedToolLine(f *toolFailure) string { + name := strings.TrimSpace(f.Tool) + if name == "" { + name = "工具" + } + reason := strings.TrimSpace(f.Summary) + if reason == "" { + return name + " 未完成（" + f.Status + "）" + } + return name + " 未完成（" + f.Status + "）：" + reason +} diff --git a/internal/runtime/squad_events_test.go b/internal/runtime/squad_events_test.go new file mode 100644 index 0000000..c545e03 --- /dev/null +++ b/internal/runtime/squad_events_test.go @@ -0,0 +1,443 @@ +package runtime + +import ( + "encoding/json" + "fmt" + "strings" + "sync" + "testing" + + agentllm "agent-native/runtime/llm" + "agent-squad/squad" + core "agent-system-core" +) + +// eventProbe 记录小队过程帧回调发射出去的事件（顺序与内容都要能断言）。 +// Emit 由 squadEventLog 的临界区内调用，因此这里无需再自锁。 +type eventProbe struct { + *entryFake + rec *squadEventLog + toolCall []core.ToolCallPayload + result []core.ToolResultPayload +} + +func newEventProbe() *eventProbe { + p := &eventProbe{entryFake: &entryFake{}} + p.rec = newSquadEventLog(p) + return p +} + +func (p *eventProbe) Emit(typ core.StreamEventType, payload any) error { + switch v := payload.(type) { + case core.ToolCallPayload: + p.toolCall = append(p.toolCall, v) + case core.ToolResultPayload: + p.result = append(p.result, v) + } + return p.entryFake.Emit(typ, payload) +} + +func (p *eventProbe) onEvent(ev squad.ProcessEvent) { p.rec.onEvent(ev) } + +func (p *eventProbe) parts() []core.Part { return p.rec.toolParts() } + +func (p *eventProbe) calls() []core.ToolCallPayload { + return append([]core.ToolCallPayload(nil), p.toolCall...) +} + +func (p *eventProbe) results() []core.ToolResultPayload { + return append([]core.ToolResultPayload(nil), p.result...) +} + +func toolCallsOf(parts []core.Part) []core.ToolCallPart { + var out []core.ToolCallPart + for _, p := range parts { + if p.ToolCall != nil { + out = append(out, *p.ToolCall) + } + } + return out +} + +func toolResultsOf(parts []core.Part) []core.ToolResultPart { + var out []core.ToolResultPart + for _, p := range parts { + if p.ToolResult != nil { + out = append(out, *p.ToolResult) + } + } + return out +} + +// TestSquadEventsPairByCallID：两个同名工具的 start/end 必须按 CallID 配对， +// 不得因名字相同而互相对调（这是并行同名工具显示错乱的根因）。 +func TestSquadEventsPairByCallID(t *testing.T) { + p := newEventProbe() + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_a", Detail: `{"q":"a"}`}) + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: "call_b", Detail: `{"q":"b"}`}) + // 交错到达：B 先结束。 + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_b", Status: squad.ToolStatusOK, Detail: "命中 B"}) + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "call_a", Status: squad.ToolStatusError, ErrorReason: "路径不存在"}) + + parts := p.parts() + calls := toolCallsOf(parts) + if len(calls) != 2 || calls[0].ID != "call_a" || calls[1].ID != "call_b" { + t.Fatalf("tool call parts = %+v", calls) + } + emittedCalls := p.calls() + if len(emittedCalls) != 2 || emittedCalls[0].CallID != "call_a" || emittedCalls[1].CallID != "call_b" { + t.Fatalf("emitted tool calls = %+v", emittedCalls) + } + results := toolResultsOf(parts) + if len(results) != 2 { + t.Fatalf("tool result parts = %+v", results) + } + if results[0].CallID != "call_b" || results[0].Status != squad.ToolStatusOK || results[0].Summary != "命中 B" { + t.Fatalf("result B = %+v", results[0]) + } + if results[1].CallID != "call_a" || results[1].Status != squad.ToolStatusError || results[1].Summary != "路径不存在" { + t.Fatalf("result A = %+v", results[1]) + } + if p.rec.unresolvedFailure() == nil || p.rec.unresolvedFailure().CallID != "call_a" { + t.Fatalf("unresolved failure = %+v", p.rec.unresolvedFailure()) + } +} + +// TestSquadEventsStatusIsTruthful：失败/取消/拒绝必须原样透传，不得一律报 ok， +// 摘要取真实原因（截断），部件与发射载荷完全一致。 +func TestSquadEventsStatusIsTruthful(t *testing.T) { + long := strings.Repeat("失败原因", 80) + cases := []struct { + name string + ev squad.ProcessEvent + status string + summary string + }{ + {"error", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c1", + Status: squad.ToolStatusError, ErrorReason: "权限不足", Detail: "旧摘要"}, squad.ToolStatusError, "权限不足"}, + {"denied", squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c2", + Status: squad.ToolStatusDenied, ErrorReason: "策略拒绝执行"}, squad.ToolStatusDenied, "策略拒绝执行"}, + {"canceled", squad.ProcessEvent{Type: "tool_execution_end", Tool: "web_search", CallID: "c3", + Status: squad.ToolStatusCanceled, ErrorReason: "用户取消"}, squad.ToolStatusCanceled, "用户取消"}, + {"ok", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c4", + Status: squad.ToolStatusOK, Detail: "命中 3 处"}, squad.ToolStatusOK, "命中 3 处"}, + // 有失败信号但状态未知：不得报 ok。 + {"status-less error signal", squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: "c5", + ErrorReason: "连接被拒绝", Detail: "旧摘要"}, squad.ToolStatusError, "连接被拒绝"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + status, summary := toolEventOutcome(tc.ev) + if status != tc.status || summary != tc.summary { + t.Fatalf("outcome = %q/%q, want %q/%q", status, summary, tc.status, tc.summary) + } + p := newEventProbe() + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: tc.ev.Tool, CallID: tc.ev.CallID}) + p.onEvent(tc.ev) + got := toolResultsOf(p.parts()) + if len(got) != 1 { + t.Fatalf("parts = %+v", p.parts()) + } + // 落库部件与发射载荷同源。 + if got[0].Status != tc.status || got[0].Summary != tc.summary || got[0].CallID != tc.ev.CallID { + t.Fatalf("result part = %+v", got[0]) + } + wantPayload := core.ToolResultPayload{CallID: tc.ev.CallID, Status: tc.status, Summary: tc.summary} + if gotPayloads := p.results(); len(gotPayloads) != 1 || gotPayloads[0] != wantPayload { + t.Fatalf("emitted payload = %+v", gotPayloads) + } + if gotCalls := p.calls(); len(gotCalls) != 1 || gotCalls[0].CallID != tc.ev.CallID || gotCalls[0].Name != tc.ev.Tool { + t.Fatalf("emitted tool call = %+v", gotCalls) + } + }) + } + t.Run("summary truncated", func(t *testing.T) { + status, summary := toolEventOutcome(squad.ProcessEvent{Status: squad.ToolStatusError, ErrorReason: long}) + if status != squad.ToolStatusError { + t.Fatalf("status = %q", status) + } + if want := truncateRun(long, toolResultSummaryLen); summary != want { + t.Fatalf("summary = %q, want truncated %q", summary, want) + } + }) +} + +// TestSquadEventsLegacyFramesUnchanged：字段全空的 legacy 事件必须与旧行为字节一致 +// （按工具名配对、状态恒为 ok、摘要取 Detail）。 +func TestSquadEventsLegacyFramesUnchanged(t *testing.T) { + p := newEventProbe() + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", Detail: `{"q":"x"}`}) + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", Detail: strings.Repeat("d", 200)}) + p.onEvent(squad.ProcessEvent{Type: "text_delta", Text: "结论"}) + p.onEvent(squad.ProcessEvent{Type: "usage", PromptTokens: 3, CompletionTokens: 1, LastPromptTokens: 2, CacheReadTokens: 1}) + + parts := p.parts() + calls := toolCallsOf(parts) + if len(calls) != 1 { + t.Fatalf("calls = %+v", calls) + } + wantCall := core.ToolCallPart{ID: "grep", Name: "grep", Arguments: `{"q":"x"}`} + if calls[0] != wantCall { + t.Fatalf("call part = %+v, want %+v", calls[0], wantCall) + } + results := toolResultsOf(parts) + wantResult := core.ToolResultPart{CallID: "grep", Status: "ok", Summary: truncateRun(strings.Repeat("d", 200), 160)} + if len(results) != 1 || results[0] != wantResult { + t.Fatalf("result part = %+v, want %+v", results, wantResult) + } + emitted := p.results() + if len(emitted) != 1 || emitted[0].CallID != "grep" || emitted[0].Status != "ok" || + emitted[0].Summary != truncateRun(strings.Repeat("d", 200), 160) { + t.Fatalf("emitted = %+v", emitted) + } + if p.rec.reply() != "结论" { + t.Fatalf("reply = %q", p.rec.reply()) + } + if f := p.rec.unresolvedFailure(); f != nil { + t.Fatalf("legacy frames must not register a failure: %+v", f) + } + if p.rec.finalReply("", "", "已完成。") != "已完成。" { + t.Fatalf("legacy fallback text changed: %q", p.rec.finalReply("", "", "已完成。")) + } +} + +// TestSquadEventsConcurrentCallbacksKeepEveryEntry：并行回调（-race）下不得丢条目、 +// 不得串字段；每次调用自己的 start/end 仍成对。 +func TestSquadEventsConcurrentCallbacksKeepEveryEntry(t *testing.T) { + const calls = 64 + p := newEventProbe() + wantStatus := map[string]string{} + var wg sync.WaitGroup + for i := 0; i < calls; i++ { + id := fmt.Sprintf("call_%02d", i) + wantStatus[id] = squad.ToolStatusOK + failing := i%8 == 0 + if failing { + wantStatus[id] = squad.ToolStatusError + } + wg.Add(1) + go func() { + defer wg.Done() + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "grep", CallID: id, Detail: id}) + if failing { + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: id, + Status: squad.ToolStatusError, ErrorReason: "boom " + id}) + return + } + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "grep", CallID: id, + Status: squad.ToolStatusOK, Detail: "ok " + id}) + }() + } + wg.Wait() + + parts := p.parts() + if got := len(toolCallsOf(parts)); got != calls { + t.Fatalf("tool call parts = %d, want %d", got, calls) + } + results := toolResultsOf(parts) + if len(results) != calls { + t.Fatalf("tool result parts = %d, want %d", len(results), calls) + } + seenStart := map[string]bool{} + for _, c := range toolCallsOf(parts) { + if seenStart[c.ID] { + t.Fatalf("duplicate start for %s", c.ID) + } + seenStart[c.ID] = true + } + emitted := map[string]string{} + for _, r := range results { + if _, dup := emitted[r.CallID]; dup { + t.Fatalf("duplicate end for %s", r.CallID) + } + emitted[r.CallID] = r.Status + if !seenStart[r.CallID] { + t.Fatalf("end without start: %s", r.CallID) + } + } + for id, status := range emitted { + if status != wantStatus[id] { + t.Fatalf("end %s status = %q, want %q", id, status, wantStatus[id]) + } + } +} + +// TestSquadEventsConcurrentToolCallbacksShareArtifactList：宿主工具被并行调用时， +// 登记的工件一条都不能丢，snapshot 也不应把内部切片直接交出去。 +func TestSquadEventsConcurrentToolCallbacksShareArtifactList(t *testing.T) { + const n = 128 + produced := &artifactList{} + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + produced.add(&core.ArtifactRef{ArtifactID: core.ID(fmt.Sprintf("art_%03d", i)), Name: "纪要.docx", Kind: "document"}) + }(i) + } + wg.Wait() + got := produced.snapshot() + if len(got) != n { + t.Fatalf("artifacts = %d, want %d", len(got), n) + } + ids := map[string]bool{} + for _, ref := range got { + ids[string(ref.ArtifactID)] = true + } + for i := 0; i < n; i++ { + if !ids[fmt.Sprintf("art_%03d", i)] { + t.Fatalf("missing artifact %d", i) + } + } + got[0] = nil // 快照必须是副本：改动不得回写内部表 + if produced.snapshot()[0] == nil { + t.Fatal("snapshot aliases the internal list") + } +} + +// TestSquadEventsFinalTextMatchesToolOutcome：未恢复的失败不得被收尾文本改写成成功； +// 同名工具重试成功后收尾文本恢复旧行为。 +func TestSquadEventsFinalTextMatchesToolOutcome(t *testing.T) { + p := newEventProbe() + p.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c1"}) + p.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c1", + Status: squad.ToolStatusCanceled, ErrorReason: "用户取消"}) + if got := p.rec.finalReply("", "", "已完成。"); got == "已完成。" || !strings.Contains(got, "run_command") || !strings.Contains(got, "用户取消") { + t.Fatalf("fallback text = %q", got) + } + if got := p.rec.finalReply("已按你的要求处理。", "", "已完成。"); got != "已按你的要求处理。" { + t.Fatalf("model text must be preserved: %q", got) + } + // 失败部件必须留在最终部件里，不被收尾文本覆盖。 + if results := toolResultsOf(p.parts()); len(results) != 1 || results[0].Status != squad.ToolStatusCanceled { + t.Fatalf("failure part dropped: %+v", results) + } + + // 同名工具重试成功 → 失败记为已恢复，兜底文本回到旧行为。 + q := newEventProbe() + q.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c1"}) + q.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c1", + Status: squad.ToolStatusError, ErrorReason: "超时"}) + q.onEvent(squad.ProcessEvent{Type: "tool_execution_start", Tool: "run_command", CallID: "c2"}) + q.onEvent(squad.ProcessEvent{Type: "tool_execution_end", Tool: "run_command", CallID: "c2", + Status: squad.ToolStatusOK, Detail: "done"}) + if f := q.rec.unresolvedFailure(); f != nil { + t.Fatalf("retry should clear the failure: %+v", f) + } + if got := q.rec.finalReply("", "", "已完成。"); got != "已完成。" { + t.Fatalf("recovered run fallback = %q", got) + } +} + +// entryProbe 在真实运行桩上记录工具事件载荷，并让指定工具真的失败， +// 用来端到端验证「工具结局 → 过程事件 → 落库部件」这条链。 +type entryProbe struct { + *entryFake + mu sync.Mutex + calls []core.ToolCallPayload + results []core.ToolResultPayload +} + +func (p *entryProbe) Emit(typ core.StreamEventType, payload any) error { + switch v := payload.(type) { + case core.ToolCallPayload: + p.mu.Lock() + p.calls = append(p.calls, v) + p.mu.Unlock() + case core.ToolResultPayload: + p.mu.Lock() + p.results = append(p.results, v) + p.mu.Unlock() + } + return p.entryFake.Emit(typ, payload) +} + +func (p *entryProbe) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult { + if call.Name == "knowledge_search" { + return core.ToolCallResult{Status: "error", Output: "索引不可用"} + } + return p.entryFake.ExecuteTool(call) +} + +func (p *entryProbe) resultPayloads() []core.ToolResultPayload { + p.mu.Lock() + defer p.mu.Unlock() + return append([]core.ToolResultPayload(nil), p.results...) +} + +// TestSquadEntryFailedToolReachesFinalParts：真实跑一次小队入口，同名并行工具 +// 必须按各自 CallID 落部件，失败必须显示为失败，且不被收尾文本改写成成功。 +func TestSquadEntryFailedToolReachesFinalParts(t *testing.T) { + dir := t.TempDir() + base := baseEntry(dir) + rc := &entryProbe{entryFake: base} + rc.tools = []core.ToolDesc{{Name: "knowledge_search", Description: "查资料", Parameters: json.RawMessage(`{"type":"object"}`)}} + + runner := &SquadRunner{ + Router: fixedRoute{mode: squad.RouteDirect}, Checker: squad.MinimalChecker{}, + Hitl: squad.HitlPolicy{Mode: squad.HitlDisabled}, + Stream: scriptedCalls(t, nil, [][]agentllm.Event{ + {{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopToolCalls, + ToolCalls: []agentllm.ToolCall{ + {ID: "call_k1", Name: "knowledge_search", Arguments: `{"q":"a"}`}, + {ID: "call_k2", Name: "knowledge_search", Arguments: `{"q":"b"}`}, + }}}}, + {{Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, StopReason: agentllm.StopToolCalls, + ToolCalls: []agentllm.ToolCall{{ID: "w", Name: "write_file", Arguments: mustDocArgs()}}}}}, + {{Kind: agentllm.KindTextDelta, Text: "写完了。"}, + {Kind: agentllm.KindMessageEnd, Message: agentllm.Message{Role: agentllm.RoleAssistant, Content: "写完了。", StopReason: agentllm.StopEnd}}}, + }), + } + if err := runner.Execute(rc); err != nil { + t.Fatal(err) + } + parts := base.parts + calls := toolCallsOf(parts) + ids := map[string]bool{} + for _, c := range calls { + if c.Name != "knowledge_search" { + continue + } + if c.ID == "knowledge_search" || ids[c.ID] { + t.Fatalf("tool call part lost its call id: %+v (all %+v)", c, calls) + } + ids[c.ID] = true + } + if len(ids) != 2 { + t.Fatalf("distinct call ids = %v", ids) + } + results := toolResultsOf(parts) + search := map[string]core.ToolResultPart{} + for _, r := range results { + if ids[r.CallID] { + search[r.CallID] = r + } + } + if len(search) != 2 { + t.Fatalf("tool results for knowledge_search = %+v", results) + } + for id, r := range search { + if r.Status != squad.ToolStatusError { + t.Fatalf("%s status = %q, want error", id, r.Status) + } + if r.Summary != "索引不可用" { + t.Fatalf("%s summary = %q", id, r.Summary) + } + } + // 发射载荷与落库部件一致：前端过程行不得说 ok。 + emitted := map[string]string{} + for _, pl := range rc.resultPayloads() { + if ids[pl.CallID] { + emitted[pl.CallID] = pl.Status + } + } + for id, r := range search { + if emitted[id] != r.Status { + t.Fatalf("emitted %s status = %q, part status = %q", id, emitted[id], r.Status) + } + } + // 收尾文本保留模型原话，失败部件不被覆盖。 + if last := parts[len(parts)-1]; last.Type != core.PartText || last.Text != "写完了。" { + t.Fatalf("final part = %+v", last) + } +} diff --git a/internal/runtime/squad_runner.go b/internal/runtime/squad_runner.go index 7a68bbf..a82d721 100644 --- a/internal/runtime/squad_runner.go +++ b/internal/runtime/squad_runner.go @@ -30,37 +30,8 @@ func (s *SquadRunner) Execute(rc RunCtx) error { prof := rc.Profile() maxTurns, maxTools := loopCaps(prof.Budget) produced := &artifactList{} - var streamed strings.Builder - var toolParts []core.Part - textIdx := 0 - onEvent := func(ev squad.ProcessEvent) { - switch ev.Type { - case "text_delta": - if ev.Text == "" { - return - } - streamed.WriteString(ev.Text) - _ = rc.Emit(core.EvMessageDelta, core.MessageDeltaPayload{Index: textIdx, Text: ev.Text}) - case "reasoning_delta": - if ev.Text != "" { - _ = rc.Emit(core.EvThinkingDelta, core.ThinkingDeltaPayload{Text: ev.Text}) - } - case "tool_execution_start": - _ = rc.Emit(core.EvToolCall, core.ToolCallPayload{CallID: ev.Tool, Name: ev.Tool, Arguments: ev.Detail}) - toolParts = append(toolParts, core.Part{Type: core.PartToolCall, Process: true, ToolCall: &core.ToolCallPart{ID: ev.Tool, Name: ev.Tool, Arguments: ev.Detail}}) - case "tool_execution_end": - _ = rc.Emit(core.EvToolResult, core.ToolResultPayload{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)}) - toolParts = append(toolParts, core.Part{Type: core.PartToolResult, Process: true, ToolResult: &core.ToolResultPart{CallID: ev.Tool, Status: "ok", Summary: truncateRun(ev.Detail, 160)}}) - case "usage": - if sink, ok := rc.(interface{ AddUsage(core.Usage) }); ok { - sink.AddUsage(core.Usage{ - PromptTokens: ev.PromptTokens, CompletionTokens: ev.CompletionTokens, TotalTokens: ev.PromptTokens + ev.CompletionTokens, - LastPromptTokens: ev.LastPromptTokens, LastCompletionTokens: ev.LastCompletionTokens, - CacheReadTokens: ev.CacheReadTokens, CacheWriteTokens: ev.CacheWriteTokens, - }) - } - } - } + // 过程帧可能来自并行的工具回调：聚合全部走 squadEventLog（内部互斥）。 + evlog := newSquadEventLog(rc) router := s.Router if router == nil { router = newModelRouter(rc) @@ -73,7 +44,7 @@ func (s *SquadRunner) Execute(rc RunCtx) error { DataDir: rc.Settings().GetString("storage.dataDir", "./data"), Goal: entryGoal(rc), HostTools: bridgeHostTools(rc, produced), - OnEvent: onEvent, + OnEvent: evlog.onEvent, Stream: stream, Router: router, Checker: s.Checker, @@ -90,14 +61,9 @@ func (s *SquadRunner) Execute(rc RunCtx) error { if err != nil { return err } - reply := strings.TrimSpace(streamed.String()) - if reply == "" { - reply = strings.TrimSpace(res.Summary) - } - if reply == "" { - reply = "已完成。" - } - if claimsFalseDeliverable(reply, produced.items, rc) { + // 收尾文本：模型文本原样保留；仍有未恢复的工具失败时不用「已完成。」兜底。 + reply := evlog.finalReply(evlog.reply(), res.Summary, "已完成。") + if claimsFalseDeliverable(reply, produced.snapshot(), rc) { return fmt.Errorf("声称已生成文件，但没有登记对应工件") } if strings.TrimSpace(res.Body) != "" { @@ -105,9 +71,9 @@ func (s *SquadRunner) Execute(rc RunCtx) error { return werr } } - parts := append([]core.Part{}, toolParts...) + parts := evlog.toolParts() parts = append(parts, core.Part{Type: core.PartText, Text: reply}) - for _, ref := range produced.items { + for _, ref := range produced.snapshot() { parts = append(parts, core.Part{Type: core.PartArtifactRef, Artifact: ref}) } return rc.SetAssistantParts(parts)