1package subagents
2
3import (
4 "context"
5 "os"
6 "path/filepath"
7 "strings"
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, MaxTotalSpawned: 3, 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 TestSchedulerRejectsChildSpawningAndTotalBudget(t *testing.T) {
99 f := New(Config{Root: t.TempDir(), RepoRoot: t.TempDir(), Policy: SubagentPolicy{MaxTotalSpawned: 1}, 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 first, err := f.Spawn(context.Background(), "one")
106 if err != nil {
107 t.Fatal(err)
108 }
109 first.Wait()
110 if _, err := f.Spawn(context.Background(), "two"); err == nil || !strings.Contains(err.Error(), "budget") {
111 t.Fatalf("second spawn error = %v, want budget rejection", err)
112 }
113}
114
115func TestStopQueuedAgentBeforeRunnerAdmission(t *testing.T) {
116 root := t.TempDir()
117 started := make(chan string, 2)
118 f := New(Config{
119 Root: root, RepoRoot: root,
120 Policy: SubagentPolicy{MaxConcurrent: 1, MaxTotalSpawned: 3, IdleTimeout: time.Hour},
121 NewRunner: func(a *Agent) Runner {
122 return RunnerFunc(func(ctx context.Context, _ Sink) error {
123 started <- a.ID
124 <-ctx.Done()
125 return ctx.Err()
126 })
127 },
128 })
129 first, err := f.Spawn(context.Background(), "first")
130 if err != nil {
131 t.Fatal(err)
132 }
133 select {
134 case got := <-started:
135 if got != first.ID {
136 t.Fatalf("first runner = %q, want %q", got, first.ID)
137 }
138 case <-time.After(time.Second):
139 t.Fatal("first runner did not start")
140 }
141 queued, err := f.Spawn(context.Background(), "queued")
142 if err != nil {
143 t.Fatal(err)
144 }
145 if err := f.Stop(queued.ID); err != nil {
146 t.Fatal(err)
147 }
148 queued.Wait()
149 if queued.Status() != StatusKilled {
150 t.Fatalf("queued status = %s, want killed", queued.Status())
151 }
152 select {
153 case got := <-started:
154 t.Fatalf("stopped queued agent still reached runner: %q", got)
155 case <-time.After(50 * time.Millisecond):
156 }
157 if err := f.Stop(first.ID); err != nil {
158 t.Fatal(err)
159 }
160 first.Wait()
161}
162
163func TestCanceledQueuedAgentIsRemovedAndGetsResult(t *testing.T) {
164 root := t.TempDir()
165 started := make(chan struct{})
166 f := New(Config{
167 Root: root, RepoRoot: root,
168 Policy: SubagentPolicy{MaxConcurrent: 1, MaxTotalSpawned: 4, IdleTimeout: time.Hour},
169 NewRunner: func(*Agent) Runner {
170 return RunnerFunc(func(ctx context.Context, _ Sink) error {
171 select {
172 case started <- struct{}{}:
173 default:
174 }
175 <-ctx.Done()
176 return ctx.Err()
177 })
178 },
179 })
180 first, err := f.Spawn(context.Background(), "first")
181 if err != nil {
182 t.Fatal(err)
183 }
184 <-started
185 queued, err := f.Spawn(context.Background(), "queued")
186 if err != nil {
187 t.Fatal(err)
188 }
189 if queued.Status() != StatusPending {
190 t.Fatalf("queued status = %s, want pending", queued.Status())
191 }
192 if err := f.Stop(queued.ID); err != nil {
193 t.Fatal(err)
194 }
195 queued.Wait()
196 f.mu.Lock()
197 queueLen := len(f.queue)
198 f.mu.Unlock()
199 if queueLen != 0 {
200 t.Fatalf("canceled queue entries = %d, want 0", queueLen)
201 }
202 result, err := f.ReadResult(queued.ID)
203 if err != nil {
204 t.Fatal(err)
205 }
206 if result.Status != ResultCanceled {
207 t.Fatalf("queued cancellation result = %s, want canceled", result.Status)
208 }
209 if err := f.Stop(first.ID); err != nil {
210 t.Fatal(err)
211 }
212 first.Wait()
213}
214
215func TestSpawnIDsRemainUniqueWithSameClock(t *testing.T) {
216 now := time.Unix(123, 0)
217 f := New(Config{
218 Root: t.TempDir(), RepoRoot: t.TempDir(), Now: func() time.Time { return now },
219 NewRunner: func(*Agent) Runner {
220 return RunnerFunc(func(context.Context, Sink) error { return nil })
221 },
222 })
223 first, err := f.Spawn(context.Background(), "same task")
224 if err != nil {
225 t.Fatal(err)
226 }
227 second, err := f.Spawn(context.Background(), "same task")
228 if err != nil {
229 t.Fatal(err)
230 }
231 first.Wait()
232 second.Wait()
233 if first.ID == second.ID {
234 t.Fatalf("duplicate agent ids: %q", first.ID)
235 }
236}
237
238func TestStructuredFallbackResultIsDurable(t *testing.T) {
239 root := t.TempDir()
240 f := New(Config{
241 Root: root, RepoRoot: root,
242 NewRunner: func(*Agent) Runner {
243 return RunnerFunc(func(context.Context, Sink) error {
244 return nil
245 })
246 },
247 })
248 a, err := f.Spawn(context.Background(), "result fallback")
249 if err != nil {
250 t.Fatal(err)
251 }
252 a.Wait()
253 result, err := f.ReadResult(a.ID)
254 if err != nil {
255 t.Fatal(err)
256 }
257 if result.Status != ResultSucceeded || result.AgentID != a.ID || result.TurnID == "" {
258 t.Fatalf("result = %+v", result)
259 }
260 if _, err := os.Stat(filepath.Join(root, "agents", a.ID, "result.json")); err != nil {
261 t.Fatalf("result file missing: %v", err)
262 }
263}