Agent-System/internal/squad/process_bridge_test.go

package squad

import (
	"strings"
	"sync"
	"testing"
	"time"

	"chat-agent-system/internal/agent"
	"chat-agent-system/internal/stream"
)

func TestProcessBridgeNilStreamerIsNoop(t *testing.T) {
	step := StepSpec{ID: "step_01", Role: "regulatory"}
	req := &agent.Request{}
	stop := AttachProcessBridge(nil, step, req)
	if req.OnEvent != nil {
		t.Fatal("nil streamer must not attach OnEvent")
	}
	req.OnEvent = func(agent.ProcessEvent) {} // 直接赋值不应有 panic
	stop()                                    // no-op stop 不 panic
}

func TestProcessBridgeReasoningFlow(t *testing.T) {
	var mu sync.Mutex
	var events []stream.StreamEvent
	streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) {
		mu.Lock()
		events = append(events, ev)
		mu.Unlock()
	})
	step := StepSpec{ID: "step_02", Role: "engineer"}
	req := &agent.Request{}
	stop := AttachProcessBridge(streamer, step, req)
	defer stop()

	req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "第一段思考"})
	req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "继续推理"})
	req.OnEvent(agent.ProcessEvent{Type: "message_end"})
	stop()
	// 再发事件（stop 后）不应 panic 且不应再发帧
	req.OnEvent(agent.ProcessEvent{Type: "reasoning_delta", Text: "ignored"})

	mu.Lock()
	defer mu.Unlock()
	var deltas, dones int
	var joined strings.Builder
	for _, ev := range events {
		if ev.Type != stream.EventDetail || ev.Detail == nil {
			continue
		}
		if ev.Detail.Category != detailCategoryAgentThought {
			t.Fatalf("unexpected category %s", ev.Detail.Category)
		}
		if ev.Detail.Delta != "" {
			deltas++
			joined.WriteString(ev.Detail.Delta)
		}
		if ev.Detail.Done {
			dones++
		}
		if ev.Detail.StepID != step.ID || ev.Detail.Role != step.Role {
			t.Fatalf("step identity missing: %+v", ev.Detail)
		}
	}
	if joined.String() != "第一段思考继续推理" {
		t.Fatalf("joined delta %q", joined.String())
	}
	if dones != 1 {
		t.Fatalf("expected exactly 1 done frame, got %d", dones)
	}
	if deltas < 1 {
		t.Fatal("expected at least one delta frame")
	}
}

func TestProcessBridgeToolFlow(t *testing.T) {
	var mu sync.Mutex
	var events []stream.StreamEvent
	streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) {
		mu.Lock()
		events = append(events, ev)
		mu.Unlock()
	})
	step := StepSpec{ID: "step_03", Role: "writer"}
	b := newProcessBridge(streamer, step, "撰写专员")
	if b == nil {
		t.Fatal("bridge expected")
	}

	b.Handle(agent.ProcessEvent{Type: "tool_call_delta", Tool: "read_file", Detail: `{"path":"a.md"`})
	b.Handle(agent.ProcessEvent{Type: "tool_execution_start", Tool: "read_file"})
	b.Handle(agent.ProcessEvent{Type: "tool_call_delta", Tool: "read_file", Detail: `}`})
	b.Handle(agent.ProcessEvent{Type: "tool_execution_end", Tool: "read_file", Detail: "file content ..."})
	b.Stop()

	mu.Lock()
	defer mu.Unlock()
	var calling, done, withInput int
	for _, ev := range events {
		if ev.Type != stream.EventDetail || ev.Detail == nil || ev.Detail.Category != detailCategoryToolCall {
			continue
		}
		if ev.Detail.Tool == nil {
			t.Fatal("tool_call frame must carry Tool info")
		}
		switch {
		case ev.Detail.Tool.Status == "calling":
			calling++
		case ev.Detail.Done:
			done++
		}
		if strings.Contains(ev.Detail.Tool.Input, "a.md") {
			withInput++
		}
	}
	if calling != 1 {
		t.Fatalf("calling frames %d", calling)
	}
	if done < 1 {
		t.Fatal("expected done frame for tool item")
	}
	if withInput == 0 {
		t.Fatal("tool input aggregate lost")
	}
}

func TestProcessBridgeToolFailureStatus(t *testing.T) {
	var mu sync.Mutex
	var events []stream.StreamEvent
	streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) {
		mu.Lock()
		events = append(events, ev)
		mu.Unlock()
	})
	b := newProcessBridge(streamer, StepSpec{ID: "s", Role: "r"}, "测试")
	b.Handle(agent.ProcessEvent{Type: "tool_execution_start", Tool: "write_file"})
	b.Handle(agent.ProcessEvent{Type: "tool_execution_end", Tool: "write_file", Detail: "error: disk full"})
	b.Stop()

	mu.Lock()
	defer mu.Unlock()
	sawFailed := false
	for _, ev := range events {
		if ev.Detail != nil && ev.Detail.Tool != nil && ev.Detail.Tool.Status == "failed" {
			sawFailed = true
		}
	}
	if !sawFailed {
		t.Fatal("error result must map to failed status")
	}
}

func TestTrimProcessText(t *testing.T) {
	small := strings.Repeat("a", 100)
	if got := trimProcessText(small); got != small {
		t.Fatal("small text must pass through")
	}
	big := strings.Repeat("b", processToolTrimBytes*4)
	got := trimProcessText(big)
	if len(got) >= len(big) {
		t.Fatal("big text must be trimmed")
	}
	if !strings.Contains(got, "…(中间省略") {
		t.Fatal("trim marker missing")
	}
}

func TestProcessBridgeConcurrentSteps(t *testing.T) {
	// 并行组成员各自持有独立 bridge，事件互不串扰（设计 F.4）。
	var mu sync.Mutex
	seen := map[string]string{} // itemID -> role
	streamer := stream.NewStreamer(true, func(ev stream.StreamEvent) {
		if ev.Type != stream.EventDetail || ev.Detail == nil {
			return
		}
		mu.Lock()
		seen[ev.Detail.ItemID] = ev.Detail.Role
		mu.Unlock()
	})
	b1 := newProcessBridge(streamer, StepSpec{ID: "p1", Role: "regulatory"}, "法规专员")
	b2 := newProcessBridge(streamer, StepSpec{ID: "p2", Role: "market"}, "市场专员")
	var wg sync.WaitGroup
	wg.Add(2)
	go func() {
		defer wg.Done()
		for i := 0; i < 20; i++ {
			b1.Handle(agent.ProcessEvent{Type: "reasoning_delta", Text: "r1"})
		}
		b1.Stop()
	}()
	go func() {
		defer wg.Done()
		for i := 0; i < 20; i++ {
			b2.Handle(agent.ProcessEvent{Type: "reasoning_delta", Text: "r2"})
		}
		b2.Stop()
	}()
	wg.Wait()
	time.Sleep(50 * time.Millisecond) // 等冲刷
	mu.Lock()
	defer mu.Unlock()
	for itemID, role := range seen {
		if !strings.Contains(itemID, "p1") && !strings.Contains(itemID, "p2") {
			t.Fatalf("unexpected item %s", itemID)
		}
		want := "regulatory"
		if strings.Contains(itemID, "p2") {
			want = "market"
		}
		if role != want {
			t.Fatalf("item %s mapped to role %s, want %s", itemID, role, want)
		}
	}
}