weichen
fc0a0505b3
Why: F14 backend half - tool events used the tool name as the call id and hardcoded status ok, so same-named tools could not be paired and failures displayed as success; parallel callbacks shared artifact state. Evidence: squad runner event tests pass (verify-u11c.json); pairing, error mapping and legacy passthrough cases added; artifact list locked. Impact: legacy empty-field events keep byte-identical output. Refs: feature/agent-stack-audit-remediation-b2 U11c
444 lines
16 KiB
Go
444 lines
16 KiB
Go
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)
|
|
}
|
|
}
|