1package subagents
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "runtime"
8 "strings"
9 "sync"
10 "testing"
11 "time"
12)
13
14// newTestSupervisor builds a Supervisor rooted in t.TempDir and configured
15// with the Runner factory controlled by the test.
16func newTestSupervisor(t *testing.T, mk func(a *Agent) Runner) *Supervisor {
17 t.Helper()
18 root := t.TempDir()
19 return New(Config{
20 Root: root,
21 RepoRoot: root,
22 NewRunner: mk,
23 })
24}
25
26func TestSpawnRunsAndCompletes(t *testing.T) {
27 ran := make(chan string, 1)
28 f := newTestSupervisor(t, func(a *Agent) Runner {
29 return RunnerFunc(func(ctx context.Context, sink Sink) error {
30 ran <- a.Task
31 sink.Activity("hello")
32 sink.Transcript("line one")
33 sink.Transcript("line two")
34 return nil
35 })
36 })
37 a, err := f.Spawn(context.Background(), "do a thing")
38 if err != nil {
39 t.Fatal(err)
40 }
41 select {
42 case got := <-ran:
43 if got != "do a thing" {
44 t.Fatalf("runner got task %q; want %q", got, "do a thing")
45 }
46 case <-time.After(time.Second):
47 t.Fatal("runner did not start")
48 }
49 a.Wait()
50 if a.Status() != StatusDone {
51 t.Fatalf("status %s; want done", a.Status())
52 }
53 if got := a.Transcript(); len(got) != 2 || got[0] != "line one" || got[1] != "line two" {
54 t.Fatalf("transcript = %q", got)
55 }
56 if !strings.Contains(a.ID, "do-a-thing") {
57 t.Fatalf("id %q missing slug", a.ID)
58 }
59 // Every agent shares the host's RepoRoot.
60 if a.Dir != f.cfg.RepoRoot {
61 t.Fatalf("dir = %q; want repo root %q", a.Dir, f.cfg.RepoRoot)
62 }
63}
64
65// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
66// every spawned agent points its cwd at the parent zut's RepoRoot.
67func TestSpawnAgentSharesRepoRoot(t *testing.T) {
68 f := newTestSupervisor(t, func(a *Agent) Runner {
69 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
70 })
71 a, err := f.Spawn(context.Background(), "share me")
72 if err != nil {
73 t.Fatal(err)
74 }
75 a.Wait()
76 if a.Dir != f.cfg.RepoRoot {
77 t.Fatalf("Dir = %q; want RepoRoot %q", a.Dir, f.cfg.RepoRoot)
78 }
79}
80
81func TestSetRepoRootAffectsSubsequentSpawns(t *testing.T) {
82 f := newTestSupervisor(t, func(a *Agent) Runner {
83 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
84 })
85 want := t.TempDir()
86 f.SetRepoRoot(want)
87 a, err := f.Spawn(context.Background(), "use new root")
88 if err != nil {
89 t.Fatal(err)
90 }
91 a.Wait()
92 if a.Dir != want {
93 t.Fatalf("Dir = %q; want updated RepoRoot %q", a.Dir, want)
94 }
95}
96
97func TestSpawnEmptyTaskFails(t *testing.T) {
98 f := newTestSupervisor(t, func(a *Agent) Runner {
99 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
100 })
101 if _, err := f.Spawn(context.Background(), " "); err == nil {
102 t.Fatal("expected error on empty task")
103 }
104}
105
106func TestRunnerErrorMarksFailed(t *testing.T) {
107 wantErr := errors.New("boom")
108 f := newTestSupervisor(t, func(a *Agent) Runner {
109 return RunnerFunc(func(ctx context.Context, sink Sink) error { return wantErr })
110 })
111 a, _ := f.Spawn(context.Background(), "explode")
112 a.Wait()
113 if a.Status() != StatusFailed {
114 t.Fatalf("status %s; want failed", a.Status())
115 }
116 if !errors.Is(a.Err(), wantErr) {
117 t.Fatalf("err = %v", a.Err())
118 }
119 if !strings.Contains(a.Activity(), "boom") {
120 t.Fatalf("activity = %q; want it to mention the error", a.Activity())
121 }
122}
123
124func TestStopCancelsRunningAgent(t *testing.T) {
125 started := make(chan struct{})
126 f := newTestSupervisor(t, func(a *Agent) Runner {
127 return RunnerFunc(func(ctx context.Context, sink Sink) error {
128 close(started)
129 <-ctx.Done()
130 return ctx.Err()
131 })
132 })
133 a, _ := f.Spawn(context.Background(), "long")
134 <-started
135 if err := f.Stop(a.ID); err != nil {
136 t.Fatal(err)
137 }
138 a.Wait()
139 if a.Status() != StatusKilled {
140 t.Fatalf("status = %s; want killed", a.Status())
141 }
142}
143
144func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
145 if runtime.GOOS == "windows" {
146 t.Skip("subagent inbox transport uses Unix-domain sockets")
147 }
148 const grace = 250 * time.Millisecond
149 started := make(chan struct{})
150 canceled := make(chan struct{})
151 f := New(Config{
152 Root: t.TempDir(),
153 RepoRoot: t.TempDir(),
154 Policy: SubagentPolicy{
155 CancelGracePeriod: grace,
156 },
157 NewRunner: func(a *Agent) Runner {
158 return RunnerFunc(func(ctx context.Context, sink Sink) error {
159 listener, err := Listen(a.InboxPath)
160 if err != nil {
161 return err
162 }
163 defer listener.Close()
164 close(started)
165 select {
166 case <-listener.Lines():
167 case <-ctx.Done():
168 close(canceled)
169 return ctx.Err()
170 }
171 <-ctx.Done()
172 close(canceled)
173 return ctx.Err()
174 })
175 },
176 })
177 a, err := f.Spawn(context.Background(), "graceful stop")
178 if err != nil {
179 t.Fatal(err)
180 }
181 select {
182 case <-started:
183 case <-time.After(time.Second):
184 t.Fatal("runner did not start")
185 }
186
187 ctx, cancel := context.WithCancel(context.Background())
188 if err := f.StopContext(ctx, a.ID); err != nil {
189 t.Fatal(err)
190 }
191 cancel()
192
193 early := time.NewTimer(grace / 2)
194 defer early.Stop()
195 select {
196 case <-canceled:
197 t.Fatal("agent canceled when StopContext caller context was canceled")
198 case <-early.C:
199 }
200
201 deadline := time.NewTimer(grace * 2)
202 defer deadline.Stop()
203 select {
204 case <-canceled:
205 case <-deadline.C:
206 t.Fatal("agent was not canceled after the grace period")
207 }
208 a.Wait()
209}
210
211func TestStopContextCancelsDetachedWaitWithoutHoldingOperationLock(t *testing.T) {
212 if runtime.GOOS == "windows" {
213 t.Skip("subagent inbox transport uses Unix-domain sockets")
214 }
215 path := filepath.Join(shortSocketDir(t), "agent.sock")
216 listener, err := Listen(path)
217 if err != nil {
218 t.Fatal(err)
219 }
220 defer listener.Close()
221
222 f := New(Config{
223 Root: t.TempDir(),
224 Policy: SubagentPolicy{
225 CancelGracePeriod: time.Hour,
226 },
227 })
228 a := &Agent{
229 ID: "detached-agent",
230 InboxPath: path,
231 inbox: NewInbox(path),
232 status: StatusDetached,
233 done: make(chan struct{}),
234 }
235 defer a.inbox.Close()
236 close(a.done)
237 f.mu.Lock()
238 f.agents[a.ID] = a
239 f.order = append(f.order, a.ID)
240 f.mu.Unlock()
241
242 ctx, cancel := context.WithCancel(context.Background())
243 defer cancel()
244 stopDone := make(chan error, 1)
245 go func() { stopDone <- f.StopContext(ctx, a.ID) }()
246
247 // The listener notification proves Stop reached its shutdown wait. The
248 // operation lock must already be available while that wait is in progress.
249 if msg := <-listener.Lines(); msg == "" {
250 t.Fatal("shutdown command was empty")
251 }
252 operationReleased := make(chan struct{})
253 go func() {
254 f.operationMu.Lock()
255 close(operationReleased)
256 f.operationMu.Unlock()
257 }()
258 <-operationReleased
259
260 cancel()
261 if err := <-stopDone; !errors.Is(err, context.Canceled) {
262 t.Fatalf("StopContext error = %v; want context.Canceled", err)
263 }
264 if got := a.Status(); got != StatusDetached {
265 t.Fatalf("status after canceled stop = %s; want detached", got)
266 }
267}
268
269func TestStopAfterDoneIsNoop(t *testing.T) {
270 f := newTestSupervisor(t, func(a *Agent) Runner {
271 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
272 })
273 a, _ := f.Spawn(context.Background(), "quick")
274 a.Wait()
275 if err := f.Stop(a.ID); err != nil {
276 t.Fatalf("stop after done: %v", err)
277 }
278 if a.Status() != StatusDone {
279 t.Fatalf("status flipped to %s", a.Status())
280 }
281}
282
283func TestGetPrefixMatch(t *testing.T) {
284 f := newTestSupervisor(t, func(a *Agent) Runner {
285 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
286 })
287 a, _ := f.Spawn(context.Background(), "alpha task")
288 a.Wait()
289 // Full id works.
290 if got := f.Get(a.ID); got != a {
291 t.Fatal("get by full id failed")
292 }
293 // Slug prefix works as long as it's unique.
294 if got := f.Get("alpha"); got != a {
295 t.Fatal("get by prefix failed")
296 }
297 // Bogus id returns nil.
298 if got := f.Get("zzz-nope"); got != nil {
299 t.Fatalf("expected nil; got %#v", got)
300 }
301}
302
303func TestRemoveRequiresTerminalState(t *testing.T) {
304 hold := make(chan struct{})
305 f := newTestSupervisor(t, func(a *Agent) Runner {
306 return RunnerFunc(func(ctx context.Context, sink Sink) error {
307 <-hold
308 return nil
309 })
310 })
311 a, _ := f.Spawn(context.Background(), "still going")
312 // Wait for run goroutine to flip to running.
313 for i := 0; i < 100 && a.Status() != StatusRunning; i++ {
314 time.Sleep(time.Millisecond)
315 }
316 if err := f.Remove(a.ID); err == nil {
317 t.Fatal("remove of running agent should fail")
318 }
319 close(hold)
320 a.Wait()
321 if err := f.Remove(a.ID); err != nil {
322 t.Fatalf("remove after done: %v", err)
323 }
324 if got := f.Get(a.ID); got != nil {
325 t.Fatal("agent still present after remove")
326 }
327}
328
329func TestSnapshotIsStableAcrossAccess(t *testing.T) {
330 var wg sync.WaitGroup
331 wg.Add(1)
332 f := newTestSupervisor(t, func(a *Agent) Runner {
333 return RunnerFunc(func(ctx context.Context, sink Sink) error {
334 defer wg.Done()
335 for i := 0; i < 200; i++ {
336 sink.Transcript("chunk")
337 sink.Activity("step")
338 }
339 return nil
340 })
341 })
342 a, _ := f.Spawn(context.Background(), "race")
343 // Hammer Snapshot while the runner is writing; the -race detector
344 // is the real assertion here.
345 stop := make(chan struct{})
346 go func() {
347 for {
348 select {
349 case <-stop:
350 return
351 default:
352 _ = a.Snapshot()
353 }
354 }
355 }()
356 wg.Wait()
357 a.Wait()
358 close(stop)
359 if a.Status() != StatusDone {
360 t.Fatalf("status = %s", a.Status())
361 }
362}
363
364func TestTruncateIsRuneAware(t *testing.T) {
365 input := "αβγδεζη"
366 if got := truncate(input, 6); got != "αβγ..." {
367 t.Fatalf("truncate(%q, 6) = %q; want %q", input, got, "αβγ...")
368 }
369 if got := truncate(input, 3); got != "..." {
370 t.Fatalf("truncate(%q, 3) = %q; want %q", input, got, "...")
371 }
372 if got := truncate(input, 2); got != ".." {
373 t.Fatalf("truncate(%q, 2) = %q; want %q", input, got, "..")
374 }
375}
376
377func TestTaskSlug(t *testing.T) {
378 cases := map[string]string{
379 "fix the login form": "fix-the-login-form",
380 " weird --- spaces!! ": "weird-spaces",
381 "": "agent",
382 "a-very-long-task-name-that-overflows": "a-very-long-task-name-th",
383 }
384 for in, want := range cases {
385 if got := taskSlug(in); got != want {
386 t.Errorf("taskSlug(%q) = %q; want %q", in, got, want)
387 }
388 }
389}
390
391func TestSnapshotAllSorted(t *testing.T) {
392 f := newTestSupervisor(t, func(a *Agent) Runner {
393 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
394 })
395 a1, _ := f.Spawn(context.Background(), "first")
396 // Force second spawn into a later nanosecond bucket.
397 time.Sleep(2 * time.Millisecond)
398 a2, _ := f.Spawn(context.Background(), "second")
399 a1.Wait()
400 a2.Wait()
401 snaps := f.SnapshotAll()
402 if len(snaps) != 2 {
403 t.Fatalf("want 2 snapshots; got %d", len(snaps))
404 }
405 if !snaps[0].Started.Before(snaps[1].Started) && !snaps[0].Started.Equal(snaps[1].Started) {
406 t.Fatal("snapshots not in spawn order")
407 }
408}