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 // The runner may start before the test reaches its receive. Buffer the
177 // one-shot readiness signal so scheduling cannot turn it into a lost wakeup.
178 started := make(chan struct{}, 1)
179 f := New(Config{
180 Root: root, RepoRoot: root,
181 Policy: SubagentPolicy{MaxConcurrent: 1, IdleTimeout: time.Hour},
182 NewRunner: func(*Agent) Runner {
183 return RunnerFunc(func(ctx context.Context, _ Sink) error {
184 select {
185 case started <- struct{}{}:
186 default:
187 }
188 <-ctx.Done()
189 return ctx.Err()
190 })
191 },
192 })
193 first, err := f.Spawn(context.Background(), "first")
194 if err != nil {
195 t.Fatal(err)
196 }
197 <-started
198 queued, err := f.Spawn(context.Background(), "queued")
199 if err != nil {
200 t.Fatal(err)
201 }
202 if queued.Status() != StatusPending {
203 t.Fatalf("queued status = %s, want pending", queued.Status())
204 }
205 if err := f.Stop(queued.ID); err != nil {
206 t.Fatal(err)
207 }
208 queued.Wait()
209 f.mu.Lock()
210 queueLen := len(f.queue)
211 f.mu.Unlock()
212 if queueLen != 0 {
213 t.Fatalf("canceled queue entries = %d, want 0", queueLen)
214 }
215 result, err := f.ReadResult(queued.ID)
216 if err != nil {
217 t.Fatal(err)
218 }
219 if result.Status != ResultCanceled {
220 t.Fatalf("queued cancellation result = %s, want canceled", result.Status)
221 }
222 if err := f.Stop(first.ID); err != nil {
223 t.Fatal(err)
224 }
225 first.Wait()
226}
227
228func TestSpawnIDsRemainUniqueWithSameClock(t *testing.T) {
229 now := time.Unix(123, 0)
230 f := New(Config{
231 Root: t.TempDir(), RepoRoot: t.TempDir(), Now: func() time.Time { return now },
232 NewRunner: func(*Agent) Runner {
233 return RunnerFunc(func(context.Context, Sink) error { return nil })
234 },
235 })
236 first, err := f.Spawn(context.Background(), "same task")
237 if err != nil {
238 t.Fatal(err)
239 }
240 second, err := f.Spawn(context.Background(), "same task")
241 if err != nil {
242 t.Fatal(err)
243 }
244 first.Wait()
245 second.Wait()
246 if first.ID == second.ID {
247 t.Fatalf("duplicate agent ids: %q", first.ID)
248 }
249}
250
251func TestStructuredFallbackResultIsDurable(t *testing.T) {
252 root := t.TempDir()
253 f := New(Config{
254 Root: root, RepoRoot: root,
255 NewRunner: func(*Agent) Runner {
256 return RunnerFunc(func(context.Context, Sink) error {
257 return nil
258 })
259 },
260 })
261 a, err := f.Spawn(context.Background(), "result fallback")
262 if err != nil {
263 t.Fatal(err)
264 }
265 a.Wait()
266 result, err := f.ReadResult(a.ID)
267 if err != nil {
268 t.Fatal(err)
269 }
270 if result.Status != ResultSucceeded || result.AgentID != a.ID || result.TurnID == "" {
271 t.Fatalf("result = %+v", result)
272 }
273 if _, err := os.Stat(filepath.Join(root, "agents", a.ID, "result.json")); err != nil {
274 t.Fatalf("result file missing: %v", err)
275 }
276}