1package subagents
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "os"
8 "path/filepath"
9 "testing"
10 "time"
11)
12
13func TestBatchIDsRemainUniqueWithFixedClock(t *testing.T) {
14 now := time.Unix(123, 0)
15 f := New(Config{
16 Root: t.TempDir(), RepoRoot: t.TempDir(), Now: func() time.Time { return now },
17 Policy: SubagentPolicy{MaxConcurrent: 2},
18 NewRunner: func(*Agent) Runner {
19 return RunnerFunc(func(context.Context, Sink) error { return nil })
20 },
21 })
22 t.Cleanup(f.StopAll)
23 first, err := f.SpawnBatch(context.Background(), BatchRequest{Tasks: []string{"one"}})
24 if err != nil {
25 t.Fatal(err)
26 }
27 second, err := f.SpawnBatch(context.Background(), BatchRequest{Tasks: []string{"two"}})
28 if err != nil {
29 t.Fatal(err)
30 }
31 first.Wait()
32 second.Wait()
33 if first.ID == second.ID {
34 t.Fatalf("duplicate batch ids: %q", first.ID)
35 }
36}
37
38func TestBatchReloadsDurableResult(t *testing.T) {
39 root := t.TempDir()
40 newSupervisor := func() *Supervisor {
41 return New(Config{
42 Root: root, RepoRoot: root,
43 NewRunner: func(*Agent) Runner {
44 return RunnerFunc(func(context.Context, Sink) error { return nil })
45 },
46 })
47 }
48 first := newSupervisor()
49 t.Cleanup(first.StopAll)
50 batch, err := first.SpawnBatch(context.Background(), BatchRequest{Context: "review", Tasks: []string{"one", "two"}})
51 if err != nil {
52 t.Fatal(err)
53 }
54 want, err := first.WaitBatch(batch.ID)
55 if err != nil {
56 t.Fatal(err)
57 }
58 if want.Status != BatchSucceeded {
59 t.Fatalf("first batch status = %s", want.Status)
60 }
61
62 second := newSupervisor()
63 t.Cleanup(second.StopAll)
64 loaded, errs := second.Reload()
65 if len(errs) != 0 {
66 t.Fatalf("reload errors: %v", errs)
67 }
68 if loaded != 2 {
69 t.Fatalf("reload loaded %d agents, want 2", loaded)
70 }
71 got, err := second.WaitBatch(batch.ID)
72 if err != nil {
73 t.Fatal(err)
74 }
75 if got.Status != BatchSucceeded || len(got.Results) != 2 {
76 t.Fatalf("reloaded batch = %#v", got)
77 }
78}
79
80func TestBatchReloadRejectsUnsafeIDs(t *testing.T) {
81 root := t.TempDir()
82 batchDir := filepath.Join(root, "batches")
83 if err := os.MkdirAll(batchDir, 0o700); err != nil {
84 t.Fatal(err)
85 }
86 data, err := json.Marshal(BatchResult{
87 Version: ProtocolVersion, BatchID: "../escape", TaskCount: 0, Status: BatchRunning,
88 })
89 if err != nil {
90 t.Fatal(err)
91 }
92 if err := os.WriteFile(filepath.Join(batchDir, "queued.json"), data, 0o600); err != nil {
93 t.Fatal(err)
94 }
95 f := New(Config{Root: root, RepoRoot: root})
96 if _, errs := f.Reload(); len(errs) == 0 {
97 t.Fatal("unsafe batch metadata was accepted")
98 }
99 if _, err := os.Stat(filepath.Join(root, "escape.json")); !os.IsNotExist(err) {
100 t.Fatalf("unsafe batch path was written: %v", err)
101 }
102}
103
104func TestBatchSpawnWaitAndCollect(t *testing.T) {
105 root := t.TempDir()
106 f := New(Config{
107 Root: root, RepoRoot: root,
108 Policy: SubagentPolicy{MaxConcurrent: 2, MaxConcurrentPerParent: 2, DefaultTimeout: 0, IdleTimeout: 0},
109 NewRunner: func(*Agent) Runner {
110 return RunnerFunc(func(context.Context, Sink) error { return nil })
111 },
112 })
113 t.Cleanup(f.StopAll)
114 batch, err := f.SpawnBatch(context.Background(), BatchRequest{Context: "review", Tasks: []string{"one", "two"}, MaxConcurrent: 1})
115 if err != nil {
116 t.Fatal(err)
117 }
118 result, err := f.WaitBatch(batch.ID)
119 if err != nil {
120 t.Fatal(err)
121 }
122 if result.Status != BatchSucceeded || len(result.ChildIDs) != 2 || len(result.Results) != 2 {
123 t.Fatalf("batch result = %#v", result)
124 }
125 if batch.Status() != BatchSucceeded {
126 t.Fatalf("batch status = %s", batch.Status())
127 }
128}
129
130func TestBatchAllowsMoreThanFormerLifetimeLimit(t *testing.T) {
131 const workersBeyondFormerLifetimeLimit = 33
132 tasks := make([]string, workersBeyondFormerLifetimeLimit)
133 for i := range tasks {
134 tasks[i] = fmt.Sprintf("worker-%d", i)
135 }
136 f := New(Config{
137 Root: t.TempDir(), RepoRoot: t.TempDir(),
138 Policy: SubagentPolicy{MaxConcurrent: 1, MaxConcurrentPerParent: 1},
139 NewRunner: func(*Agent) Runner {
140 return RunnerFunc(func(context.Context, Sink) error { return nil })
141 },
142 })
143 t.Cleanup(f.StopAll)
144
145 batch, err := f.SpawnBatch(context.Background(), BatchRequest{Tasks: tasks, MaxConcurrent: 1})
146 if err != nil {
147 t.Fatal(err)
148 }
149 result, err := f.WaitBatch(batch.ID)
150 if err != nil {
151 t.Fatal(err)
152 }
153 if result.Status != BatchSucceeded || len(result.ChildIDs) != len(tasks) || len(result.Results) != len(tasks) {
154 t.Fatalf("batch result = %#v", result)
155 }
156}