1package subagents
2
3import (
4 "context"
5 "fmt"
6 "os"
7 "path/filepath"
8 "sync/atomic"
9 "testing"
10 "time"
11)
12
13func TestPathWithinResolvesMissingDescendantThroughSymlink(t *testing.T) {
14 realRoot := t.TempDir()
15 aliasParent := filepath.Join(t.TempDir(), "alias")
16 if err := os.Symlink(realRoot, aliasParent); err != nil {
17 t.Skipf("directory symlinks unavailable: %v", err)
18 }
19 stateDir := filepath.Join(aliasParent, "state")
20 if err := os.MkdirAll(stateDir, 0o700); err != nil {
21 t.Fatal(err)
22 }
23 missingFile := filepath.Join(stateDir, "events.jsonl")
24 if !pathWithin(missingFile, stateDir) {
25 t.Fatalf("pathWithin(%q, %q) = false; want true", missingFile, stateDir)
26 }
27}
28
29func TestSchedulerBoundsConcurrencyAndReleasesCapacity(t *testing.T) {
30 root := t.TempDir()
31 started := make(chan string, 3)
32 var active, maxActive int32
33 f := New(Config{
34 Root: root, RepoRoot: root,
35 Policy: SubagentPolicy{MaxConcurrent: 1, MaxConcurrentPerParent: 4, QueueTimeout: time.Minute, DefaultTimeout: 0, IdleTimeout: time.Hour},
36 NewRunner: func(a *Agent) Runner {
37 return RunnerFunc(func(ctx context.Context, _ Sink) error {
38 current := atomic.AddInt32(&active, 1)
39 for {
40 old := atomic.LoadInt32(&maxActive)
41 if current <= old || atomic.CompareAndSwapInt32(&maxActive, old, current) {
42 break
43 }
44 }
45 started <- a.ID
46 <-ctx.Done()
47 atomic.AddInt32(&active, -1)
48 return ctx.Err()
49 })
50 },
51 })
52
53 first, err := f.Spawn(context.Background(), "first")
54 if err != nil {
55 t.Fatal(err)
56 }
57 second, err := f.Spawn(context.Background(), "second")
58 if err != nil {
59 t.Fatal(err)
60 }
61 third, err := f.Spawn(context.Background(), "third")
62 if err != nil {
63 t.Fatal(err)
64 }
65 select {
66 case got := <-started:
67 if got != first.ID {
68 t.Fatalf("first started = %q, want %q", got, first.ID)
69 }
70 case <-time.After(time.Second):
71 t.Fatal("first worker did not start")
72 }
73 if second.Status() != StatusPending || third.Status() != StatusPending {
74 t.Fatalf("queued statuses = %s/%s", second.Status(), third.Status())
75 }
76 if got := atomic.LoadInt32(&maxActive); got > 1 {
77 t.Fatalf("max active = %d, want <= 1", got)
78 }
79
80 if err := f.Stop(first.ID); err != nil {
81 t.Fatal(err)
82 }
83 first.Wait()
84 select {
85 case got := <-started:
86 if got != second.ID {
87 t.Fatalf("second started = %q, want %q", got, second.ID)
88 }
89 case <-time.After(time.Second):
90 t.Fatal("scheduler did not release capacity")
91 }
92 f.StopAll()
93 if got := atomic.LoadInt32(&maxActive); got > 1 {
94 t.Fatalf("max active = %d after release, want <= 1", got)
95 }
96}
97
98func TestSchedulerRejectsChildSpawning(t *testing.T) {
99 f := New(Config{Root: t.TempDir(), RepoRoot: t.TempDir(), NewRunner: func(*Agent) Runner {
100 return RunnerFunc(func(context.Context, Sink) error { return nil })
101 }})
102 if _, err := f.SpawnReq(context.Background(), SpawnRequest{Task: "child", RequesterAgentID: "worker-1"}); err == nil {
103 t.Fatal("child-originated spawn was accepted")
104 }
105}
106
107func TestSchedulerAllowsWorkersAfterManyCompletions(t *testing.T) {
108 f := New(Config{Root: t.TempDir(), RepoRoot: t.TempDir(), Policy: SubagentPolicy{MaxConcurrent: 1}, NewRunner: func(*Agent) Runner {
109 return RunnerFunc(func(context.Context, Sink) error { return nil })
110 }})
111 t.Cleanup(f.StopAll)
112
113 const workersBeyondFormerLifetimeLimit = 33
114 for worker := 0; worker < workersBeyondFormerLifetimeLimit; worker++ {
115 a, err := f.Spawn(context.Background(), fmt.Sprintf("worker-%d", worker))
116 if err != nil {
117 t.Fatalf("Spawn(worker-%d): %v", worker, err)
118 }
119 a.Wait()
120 if got := a.Status(); got != StatusDone {
121 t.Fatalf("worker-%d status = %s, want %s", worker, got, StatusDone)
122 }
123 }
124}
125
126func TestStopQueuedAgentBeforeRunnerAdmission(t *testing.T) {
127 root := t.TempDir()
128 started := make(chan string, 2)
129 f := New(Config{
130 Root: root, RepoRoot: root,
131 Policy: SubagentPolicy{MaxConcurrent: 1, IdleTimeout: time.Hour},
132 NewRunner: func(a *Agent) Runner {
133 return RunnerFunc(func(ctx context.Context, _ Sink) error {
134 started <- a.ID
135 <-ctx.Done()
136 return ctx.Err()
137 })
138 },
139 })
140 first, err := f.Spawn(context.Background(), "first")
141 if err != nil {
142 t.Fatal(err)
143 }
144 select {
145 case got := <-started:
146 if got != first.ID {
147 t.Fatalf("first runner = %q, want %q", got, first.ID)
148 }
149 case <-time.After(time.Second):
150 t.Fatal("first runner did not start")
151 }
152 queued, err := f.Spawn(context.Background(), "queued")
153 if err != nil {
154 t.Fatal(err)
155 }
156 if err := f.Stop(queued.ID); err != nil {
157 t.Fatal(err)
158 }
159 queued.Wait()
160 if queued.Status() != StatusKilled {
161 t.Fatalf("queued status = %s, want killed", queued.Status())
162 }
163 select {
164 case got := <-started:
165 t.Fatalf("stopped queued agent still reached runner: %q", got)
166 case <-time.After(50 * time.Millisecond):
167 }
168 if err := f.Stop(first.ID); err != nil {
169 t.Fatal(err)
170 }
171 first.Wait()
172}
173
174func TestCanceledQueuedAgentIsRemovedAndGetsResult(t *testing.T) {
175 root := t.TempDir()
176 started := make(chan struct{})
177 f := New(Config{
178 Root: root, RepoRoot: root,
179 Policy: SubagentPolicy{MaxConcurrent: 1, IdleTimeout: time.Hour},
180 NewRunner: func(*Agent) Runner {
181 return RunnerFunc(func(ctx context.Context, _ Sink) error {
182 select {
183 case started <- struct{}{}:
184 default:
185 }
186 <-ctx.Done()
187 return ctx.Err()
188 })
189 },
190 })
191 first, err := f.Spawn(context.Background(), "first")
192 if err != nil {
193 t.Fatal(err)
194 }
195 <-started
196 queued, err := f.Spawn(context.Background(), "queued")
197 if err != nil {
198 t.Fatal(err)
199 }
200 if queued.Status() != StatusPending {
201 t.Fatalf("queued status = %s, want pending", queued.Status())
202 }
203 if err := f.Stop(queued.ID); err != nil {
204 t.Fatal(err)
205 }
206 queued.Wait()
207 f.mu.Lock()
208 queueLen := len(f.queue)
209 f.mu.Unlock()
210 if queueLen != 0 {
211 t.Fatalf("canceled queue entries = %d, want 0", queueLen)
212 }
213 result, err := f.ReadResult(queued.ID)
214 if err != nil {
215 t.Fatal(err)
216 }
217 if result.Status != ResultCanceled {
218 t.Fatalf("queued cancellation result = %s, want canceled", result.Status)
219 }
220 if err := f.Stop(first.ID); err != nil {
221 t.Fatal(err)
222 }
223 first.Wait()
224}
225
226func TestSpawnIDsRemainUniqueWithSameClock(t *testing.T) {
227 now := time.Unix(123, 0)
228 f := New(Config{
229 Root: t.TempDir(), RepoRoot: t.TempDir(), Now: func() time.Time { return now },
230 NewRunner: func(*Agent) Runner {
231 return RunnerFunc(func(context.Context, Sink) error { return nil })
232 },
233 })
234 first, err := f.Spawn(context.Background(), "same task")
235 if err != nil {
236 t.Fatal(err)
237 }
238 second, err := f.Spawn(context.Background(), "same task")
239 if err != nil {
240 t.Fatal(err)
241 }
242 first.Wait()
243 second.Wait()
244 if first.ID == second.ID {
245 t.Fatalf("duplicate agent ids: %q", first.ID)
246 }
247}
248
249func TestStructuredFallbackResultIsDurable(t *testing.T) {
250 root := t.TempDir()
251 f := New(Config{
252 Root: root, RepoRoot: root,
253 NewRunner: func(*Agent) Runner {
254 return RunnerFunc(func(context.Context, Sink) error {
255 return nil
256 })
257 },
258 })
259 a, err := f.Spawn(context.Background(), "result fallback")
260 if err != nil {
261 t.Fatal(err)
262 }
263 a.Wait()
264 result, err := f.ReadResult(a.ID)
265 if err != nil {
266 t.Fatal(err)
267 }
268 if result.Status != ResultSucceeded || result.AgentID != a.ID || result.TurnID == "" {
269 t.Fatalf("result = %+v", result)
270 }
271 if _, err := os.Stat(filepath.Join(root, "agents", a.ID, "result.json")); err != nil {
272 t.Fatalf("result file missing: %v", err)
273 }
274}