返回 DeepSeek-Reasonix
coordinator_concurrency_test.go
根目录 / internal / agent / coordinator_concurrency_test.go
1 package agent
2
3 import (
4 "context"
5 "sync"
6 "testing"
7
8 "reasonix/internal/event"
9 "reasonix/internal/provider"
10 "reasonix/internal/tool"
11 )
12
13 type concurrentPlannerProvider struct {
14 mu sync.Mutex
15 calls int
16 }
17
18 func (p *concurrentPlannerProvider) Name() string { return "concurrent-planner" }
19
20 func (p *concurrentPlannerProvider) ContextBudgetPolicy() provider.ContextBudgetPolicy {
21 return provider.ContextBudgetPolicy{WindowMode: provider.ContextWindowIndependent}
22 }
23
24 func (p *concurrentPlannerProvider) Stream(_ context.Context, _ provider.Request) (<-chan provider.Chunk, error) {
25 p.mu.Lock()
26 p.calls++
27 p.mu.Unlock()
28 out := make(chan provider.Chunk, 2)
29 out <- provider.Chunk{Type: provider.ChunkToolCall, ToolCall: &provider.ToolCall{ID: "call-1", Name: "submit_plan", Arguments: `{"objective":"plan","steps":[{"title":"step"}]}`}}
30 out <- provider.Chunk{Type: provider.ChunkDone}
31 close(out)
32 return out, nil
33 }
34
35 func TestCoordinatorSerializesConcurrentPlannerCalls(t *testing.T) {
36 planner := &concurrentPlannerProvider{}
37 coord := NewCoordinator(planner, NewSession("planner-sys"), nil, PlannerToolRegistry(tool.NewRegistry()), Options{}, nil, 0, event.Discard, nil)
38
39 errs := make(chan error, 2)
40 for _, input := range []string{"first", "second"} {
41 go func(input string) {
42 _, err := coord.plan(context.Background(), input)
43 errs <- err
44 }(input)
45 }
46 for range 2 {
47 err := <-errs
48 if err != nil {
49 t.Fatalf("concurrent planner call: %v", err)
50 }
51 }
52 planner.mu.Lock()
53 calls := planner.calls
54 planner.mu.Unlock()
55 if calls != 2 {
56 t.Fatalf("planner calls = %d, want 2", calls)
57 }
58 }
59
59 lines GO