返回 DeepSeek-Reasonix
parallel_cancel_test.go
根目录 / internal / agent / parallel_cancel_test.go
1 package agent
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "strings"
8 "sync"
9 "testing"
10 "time"
11
12 "reasonix/internal/provider"
13 "reasonix/internal/tool"
14 )
15
16 // stubbornTool ignores its context: it returns only when released.
17 type stubbornTool struct {
18 once *sync.Once
19 started chan struct{}
20 release chan struct{}
21 }
22
23 func (stubbornTool) Name() string { return "stubborn" }
24 func (stubbornTool) Description() string { return "ignores cancellation" }
25 func (stubbornTool) Schema() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
26 func (stubbornTool) ReadOnly() bool { return true }
27 func (s stubbornTool) Execute(context.Context, json.RawMessage) (string, error) {
28 s.once.Do(func() { close(s.started) })
29 <-s.release
30 return "late", nil
31 }
32
33 // fastTool reports when it has entered execution, so the test cancels only
34 // after both tools of the batch are running.
35 type fastTool struct {
36 once *sync.Once
37 started chan struct{}
38 }
39
40 func (fastTool) Name() string { return "fast" }
41 func (fastTool) Description() string { return "always succeeds" }
42 func (fastTool) Schema() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
43 func (fastTool) ReadOnly() bool { return true }
44 func (f fastTool) Execute(context.Context, json.RawMessage) (string, error) {
45 f.once.Do(func() { close(f.started) })
46 return "ok", nil
47 }
48
49 // A read-only parallel segment must not keep the whole turn wedged behind one
50 // tool that ignores cancellation: after the grace the batch reports that call
51 // as an unknown effect while the calls that did finish keep their results.
52 func TestParallelBatchAbandonsToolThatIgnoresCancellation(t *testing.T) {
53 oldGrace := parallelStragglerGrace
54 parallelStragglerGrace = 200 * time.Millisecond
55 t.Cleanup(func() { parallelStragglerGrace = oldGrace })
56
57 stub := stubbornTool{once: &sync.Once{}, started: make(chan struct{}), release: make(chan struct{})}
58 t.Cleanup(func() { close(stub.release) })
59 fast := fastTool{once: &sync.Once{}, started: make(chan struct{})}
60 reg := tool.NewRegistry()
61 reg.Add(stub)
62 reg.Add(fast)
63 prov := &scriptedProvider{name: "p", turns: [][]provider.Chunk{
64 {toolCallChunk("stubborn-1", "stubborn", `{}`), toolCallChunk("fast-1", "fast", `{}`)},
65 {{Type: provider.ChunkText, Text: "done"}},
66 }}
67 sess := NewSession("")
68 a := New(prov, reg, sess, Options{}, &recordSink{})
69
70 ctx, cancel := context.WithCancel(context.Background())
71 defer cancel()
72 done := make(chan error, 1)
73 go func() { done <- a.Run(withNoClosedLoop(ctx), "go") }()
74 select {
75 case <-stub.started:
76 case <-time.After(5 * time.Second):
77 t.Fatal("stubborn tool never started")
78 }
79 select {
80 case <-fast.started:
81 case <-time.After(5 * time.Second):
82 t.Fatal("fast tool never started")
83 }
84 cancel()
85 select {
86 case err := <-done:
87 if !errors.Is(err, context.Canceled) {
88 t.Fatalf("Run returned %v, want context.Canceled", err)
89 }
90 case <-time.After(5 * time.Second):
91 t.Fatal("cancelled batch stayed wedged behind a tool that ignores its context")
92 }
93 if got := toolResultByID(sess, "stubborn-1"); !strings.Contains(got, "did not stop after cancellation") {
94 t.Fatalf("stubborn result = %q, want the abandoned marker", got)
95 }
96 if got := toolResultByID(sess, "fast-1"); !strings.Contains(got, "ok") {
97 t.Fatalf("fast result = %q, want the finished tool's own output", got)
98 }
99 }
100
100 lines GO