1package subagents
2
3import (
4 "context"
5 "errors"
6 "net"
7 "os"
8 "path/filepath"
9 "runtime"
10 "strings"
11 "sync"
12 "testing"
13 "time"
14)
15
16// newTestSupervisor builds a Supervisor rooted in t.TempDir and configured
17// with the Runner factory controlled by the test.
18func newTestSupervisor(t *testing.T, mk func(a *Agent) Runner) *Supervisor {
19 t.Helper()
20 root := t.TempDir()
21 return New(Config{
22 Root: root,
23 RepoRoot: root,
24 NewRunner: mk,
25 })
26}
27
28func TestUnqualifiedModelID(t *testing.T) {
29 for _, tc := range []struct {
30 name string
31 provider string
32 model string
33 want string
34 }{
35 {name: "matching provider", provider: "openai-codex", model: "openai-codex/gpt-5.6-sol", want: "gpt-5.6-sol"},
36 {name: "matching provider with whitespace", provider: "openai-codex", model: " openai-codex/gpt-5.6-sol ", want: "gpt-5.6-sol"},
37 {name: "other slash-containing model ID", provider: "openrouter", model: "anthropic/claude-sonnet-4-5", want: "anthropic/claude-sonnet-4-5"},
38 {name: "no provider", model: "/custom-model", want: "/custom-model"},
39 } {
40 t.Run(tc.name, func(t *testing.T) {
41 if got := unqualifiedModelID(tc.provider, tc.model); got != tc.want {
42 t.Fatalf("unqualifiedModelID(%q, %q) = %q, want %q", tc.provider, tc.model, got, tc.want)
43 }
44 })
45 }
46}
47
48func TestSpawnRunsAndCompletes(t *testing.T) {
49 ran := make(chan string, 1)
50 f := newTestSupervisor(t, func(a *Agent) Runner {
51 return RunnerFunc(func(ctx context.Context, sink Sink) error {
52 ran <- a.Task
53 sink.Activity("hello")
54 sink.Transcript("line one")
55 sink.Transcript("line two")
56 return nil
57 })
58 })
59 a, err := f.Spawn(context.Background(), "do a thing")
60 if err != nil {
61 t.Fatal(err)
62 }
63 select {
64 case got := <-ran:
65 if got != "do a thing" {
66 t.Fatalf("runner got task %q; want %q", got, "do a thing")
67 }
68 case <-time.After(time.Second):
69 t.Fatal("runner did not start")
70 }
71 a.Wait()
72 if a.Status() != StatusDone {
73 t.Fatalf("status %s; want done", a.Status())
74 }
75 if got := a.Transcript(); len(got) != 2 || got[0] != "line one" || got[1] != "line two" {
76 t.Fatalf("transcript = %q", got)
77 }
78 if !strings.Contains(a.ID, "do-a-thing") {
79 t.Fatalf("id %q missing slug", a.ID)
80 }
81 // Every agent shares the host's RepoRoot.
82 if a.Dir != f.cfg.RepoRoot {
83 t.Fatalf("dir = %q; want repo root %q", a.Dir, f.cfg.RepoRoot)
84 }
85}
86
87// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
88// every spawned agent points its cwd at the parent zut's RepoRoot.
89func TestSpawnAgentSharesRepoRoot(t *testing.T) {
90 f := newTestSupervisor(t, func(a *Agent) Runner {
91 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
92 })
93 a, err := f.Spawn(context.Background(), "share me")
94 if err != nil {
95 t.Fatal(err)
96 }
97 a.Wait()
98 if a.Dir != f.cfg.RepoRoot {
99 t.Fatalf("Dir = %q; want RepoRoot %q", a.Dir, f.cfg.RepoRoot)
100 }
101}
102
103func TestSetRepoRootAffectsSubsequentSpawns(t *testing.T) {
104 f := newTestSupervisor(t, func(a *Agent) Runner {
105 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
106 })
107 want := t.TempDir()
108 f.SetRepoRoot(want)
109 a, err := f.Spawn(context.Background(), "use new root")
110 if err != nil {
111 t.Fatal(err)
112 }
113 a.Wait()
114 if a.Dir != want {
115 t.Fatalf("Dir = %q; want updated RepoRoot %q", a.Dir, want)
116 }
117}
118
119func TestSpawnEmptyTaskFails(t *testing.T) {
120 f := newTestSupervisor(t, func(a *Agent) Runner {
121 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
122 })
123 if _, err := f.Spawn(context.Background(), " "); err == nil {
124 t.Fatal("expected error on empty task")
125 }
126}
127
128func TestRunnerErrorMarksFailed(t *testing.T) {
129 wantErr := errors.New("boom")
130 f := newTestSupervisor(t, func(a *Agent) Runner {
131 return RunnerFunc(func(ctx context.Context, sink Sink) error { return wantErr })
132 })
133 a, _ := f.Spawn(context.Background(), "explode")
134 a.Wait()
135 if a.Status() != StatusFailed {
136 t.Fatalf("status %s; want failed", a.Status())
137 }
138 if !errors.Is(a.Err(), wantErr) {
139 t.Fatalf("err = %v", a.Err())
140 }
141 if !strings.Contains(a.Activity(), "boom") {
142 t.Fatalf("activity = %q; want it to mention the error", a.Activity())
143 }
144}
145
146func TestStopCancelsRunningAgent(t *testing.T) {
147 started := make(chan struct{})
148 f := newTestSupervisor(t, func(a *Agent) Runner {
149 return RunnerFunc(func(ctx context.Context, sink Sink) error {
150 close(started)
151 <-ctx.Done()
152 return ctx.Err()
153 })
154 })
155 a, _ := f.Spawn(context.Background(), "long")
156 <-started
157 if err := f.Stop(a.ID); err != nil {
158 t.Fatal(err)
159 }
160 a.Wait()
161 if a.Status() != StatusKilled {
162 t.Fatalf("status = %s; want killed", a.Status())
163 }
164}
165
166func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
167 if runtime.GOOS == "windows" {
168 t.Skip("subagent inbox transport uses Unix-domain sockets")
169 }
170 const grace = 250 * time.Millisecond
171 started := make(chan struct{})
172 canceled := make(chan struct{})
173 f := New(Config{
174 Root: t.TempDir(),
175 RepoRoot: t.TempDir(),
176 Policy: SubagentPolicy{
177 CancelGracePeriod: grace,
178 },
179 NewRunner: func(a *Agent) Runner {
180 return RunnerFunc(func(ctx context.Context, sink Sink) error {
181 listener, err := Listen(a.InboxPath)
182 if err != nil {
183 return err
184 }
185 defer listener.Close()
186 close(started)
187 select {
188 case <-listener.Lines():
189 case <-ctx.Done():
190 close(canceled)
191 return ctx.Err()
192 }
193 <-ctx.Done()
194 close(canceled)
195 return ctx.Err()
196 })
197 },
198 })
199 a, err := f.Spawn(context.Background(), "graceful stop")
200 if err != nil {
201 t.Fatal(err)
202 }
203 select {
204 case <-started:
205 case <-time.After(time.Second):
206 t.Fatal("runner did not start")
207 }
208
209 ctx, cancel := context.WithCancel(context.Background())
210 if err := f.StopContext(ctx, a.ID); err != nil {
211 t.Fatal(err)
212 }
213 cancel()
214
215 early := time.NewTimer(grace / 2)
216 defer early.Stop()
217 select {
218 case <-canceled:
219 t.Fatal("agent canceled when StopContext caller context was canceled")
220 case <-early.C:
221 }
222
223 deadline := time.NewTimer(grace * 2)
224 defer deadline.Stop()
225 select {
226 case <-canceled:
227 case <-deadline.C:
228 t.Fatal("agent was not canceled after the grace period")
229 }
230 a.Wait()
231}
232
233func TestStopContextCancelsDetachedWaitWithoutHoldingOperationLock(t *testing.T) {
234 if runtime.GOOS == "windows" {
235 t.Skip("subagent inbox transport uses Unix-domain sockets")
236 }
237 path := filepath.Join(shortSocketDir(t), "agent.sock")
238 listener, err := Listen(path)
239 if err != nil {
240 t.Fatal(err)
241 }
242 defer listener.Close()
243
244 f := New(Config{
245 Root: t.TempDir(),
246 Policy: SubagentPolicy{
247 CancelGracePeriod: time.Hour,
248 },
249 })
250 a := &Agent{
251 ID: "detached-agent",
252 InboxPath: path,
253 inbox: NewInbox(path),
254 status: StatusDetached,
255 done: make(chan struct{}),
256 }
257 defer a.inbox.Close()
258 close(a.done)
259 f.mu.Lock()
260 f.agents[a.ID] = a
261 f.order = append(f.order, a.ID)
262 f.mu.Unlock()
263
264 ctx, cancel := context.WithCancel(context.Background())
265 defer cancel()
266 stopDone := make(chan error, 1)
267 go func() { stopDone <- f.StopContext(ctx, a.ID) }()
268
269 // The listener notification proves Stop reached its shutdown wait. The
270 // operation lock must already be available while that wait is in progress.
271 if msg := <-listener.Lines(); msg == "" {
272 t.Fatal("shutdown command was empty")
273 }
274 operationReleased := make(chan struct{})
275 go func() {
276 f.operationMu.Lock()
277 close(operationReleased)
278 f.operationMu.Unlock()
279 }()
280 <-operationReleased
281
282 cancel()
283 if err := <-stopDone; !errors.Is(err, context.Canceled) {
284 t.Fatalf("StopContext error = %v; want context.Canceled", err)
285 }
286 if got := a.Status(); got != StatusDetached {
287 t.Fatalf("status after canceled stop = %s; want detached", got)
288 }
289}
290
291func TestStopDetachedWorkerClosesWaiters(t *testing.T) {
292 if runtime.GOOS == "windows" {
293 t.Skip("subagent inbox transport uses Unix-domain sockets")
294 }
295 path := filepath.Join(shortSocketDir(t), "agent.sock")
296 listener, err := Listen(path)
297 if err != nil {
298 t.Fatal(err)
299 }
300 t.Cleanup(func() { _ = listener.Close() })
301
302 root := t.TempDir()
303 f := New(Config{Root: root, RepoRoot: root})
304 stateDir := f.agentStateDir("detached-agent")
305 if err := os.MkdirAll(stateDir, 0o700); err != nil {
306 t.Fatal(err)
307 }
308 a := &Agent{
309 ID: "detached-agent",
310 stateDir: stateDir,
311 InboxPath: path,
312 inbox: NewInbox(path),
313 status: StatusDetached,
314 done: make(chan struct{}),
315 }
316 defer a.inbox.Close()
317 f.mu.Lock()
318 f.agents[a.ID] = a
319 f.order = append(f.order, a.ID)
320 f.mu.Unlock()
321
322 stopped := make(chan error, 1)
323 go func() { stopped <- f.StopContext(context.Background(), a.ID) }()
324 select {
325 case msg := <-listener.Lines():
326 if msg == "" {
327 t.Fatal("shutdown command was empty")
328 }
329 case <-time.After(time.Second):
330 t.Fatal("detached worker did not receive shutdown")
331 }
332 if err := listener.Close(); err != nil {
333 t.Fatal(err)
334 }
335 select {
336 case err := <-stopped:
337 if err != nil {
338 t.Fatalf("stop detached worker: %v", err)
339 }
340 case <-time.After(time.Second):
341 t.Fatal("detached worker stop did not finish")
342 }
343 select {
344 case <-a.done:
345 case <-time.After(time.Second):
346 t.Fatal("detached worker stop did not close waiters")
347 }
348 result, err := readTurnResult(stateDir)
349 if err != nil {
350 t.Fatal(err)
351 }
352 if result.AgentID != a.ID || result.Status != ResultCanceled {
353 t.Fatalf("detached worker result = agent %q status %q, want agent %q status %q", result.AgentID, result.Status, a.ID, ResultCanceled)
354 }
355}
356
357func TestStopDetachedWorkerHonorsCanceledShutdownSend(t *testing.T) {
358 if runtime.GOOS == "windows" {
359 t.Skip("subagent inbox transport uses Unix-domain sockets")
360 }
361 path := filepath.Join(shortSocketDir(t), "agent.sock")
362 listener, err := Listen(path)
363 if err != nil {
364 t.Fatal(err)
365 }
366 defer listener.Close()
367
368 local, remote := net.Pipe()
369 defer remote.Close()
370 conn := &blockingWriteConn{
371 Conn: local,
372 started: make(chan struct{}),
373 release: make(chan struct{}),
374 }
375 inbox := NewInbox(path)
376 inbox.conn = conn
377 defer inbox.Close()
378
379 f := New(Config{Root: t.TempDir()})
380 a := &Agent{
381 ID: "detached-cancel",
382 InboxPath: path,
383 inbox: inbox,
384 status: StatusDetached,
385 done: make(chan struct{}),
386 }
387 f.mu.Lock()
388 f.agents[a.ID] = a
389 f.order = append(f.order, a.ID)
390 f.mu.Unlock()
391
392 ctx, cancel := context.WithCancel(context.Background())
393 stopped := make(chan error, 1)
394 go func() { stopped <- f.StopContext(ctx, a.ID) }()
395 select {
396 case <-conn.started:
397 case <-time.After(time.Second):
398 t.Fatal("detached worker shutdown send did not start")
399 }
400 cancel()
401 select {
402 case err := <-stopped:
403 if !errors.Is(err, context.Canceled) {
404 t.Fatalf("stop detached worker error = %v, want context canceled", err)
405 }
406 case <-time.After(time.Second):
407 t.Fatal("detached worker shutdown send did not honor cancellation")
408 }
409}
410
411func TestStopAfterDoneIsNoop(t *testing.T) {
412 f := newTestSupervisor(t, func(a *Agent) Runner {
413 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
414 })
415 a, _ := f.Spawn(context.Background(), "quick")
416 a.Wait()
417 if err := f.Stop(a.ID); err != nil {
418 t.Fatalf("stop after done: %v", err)
419 }
420 if a.Status() != StatusDone {
421 t.Fatalf("status flipped to %s", a.Status())
422 }
423}
424
425func TestGetPrefixMatch(t *testing.T) {
426 f := newTestSupervisor(t, func(a *Agent) Runner {
427 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
428 })
429 a, _ := f.Spawn(context.Background(), "alpha task")
430 a.Wait()
431 // Full id works.
432 if got := f.Get(a.ID); got != a {
433 t.Fatal("get by full id failed")
434 }
435 // Slug prefix works as long as it's unique.
436 if got := f.Get("alpha"); got != a {
437 t.Fatal("get by prefix failed")
438 }
439 // Bogus id returns nil.
440 if got := f.Get("zzz-nope"); got != nil {
441 t.Fatalf("expected nil; got %#v", got)
442 }
443}
444
445func TestRemoveRequiresTerminalState(t *testing.T) {
446 hold := make(chan struct{})
447 f := newTestSupervisor(t, func(a *Agent) Runner {
448 return RunnerFunc(func(ctx context.Context, sink Sink) error {
449 <-hold
450 return nil
451 })
452 })
453 a, _ := f.Spawn(context.Background(), "still going")
454 // Wait for run goroutine to flip to running.
455 for i := 0; i < 100 && a.Status() != StatusRunning; i++ {
456 time.Sleep(time.Millisecond)
457 }
458 if err := f.Remove(a.ID); err == nil {
459 t.Fatal("remove of running agent should fail")
460 }
461 close(hold)
462 a.Wait()
463 if err := f.Remove(a.ID); err != nil {
464 t.Fatalf("remove after done: %v", err)
465 }
466 if got := f.Get(a.ID); got != nil {
467 t.Fatal("agent still present after remove")
468 }
469}
470
471func TestSnapshotIsStableAcrossAccess(t *testing.T) {
472 var wg sync.WaitGroup
473 wg.Add(1)
474 f := newTestSupervisor(t, func(a *Agent) Runner {
475 return RunnerFunc(func(ctx context.Context, sink Sink) error {
476 defer wg.Done()
477 for i := 0; i < 200; i++ {
478 sink.Transcript("chunk")
479 sink.Activity("step")
480 }
481 return nil
482 })
483 })
484 a, _ := f.Spawn(context.Background(), "race")
485 // Hammer Snapshot while the runner is writing; the -race detector
486 // is the real assertion here.
487 stop := make(chan struct{})
488 go func() {
489 for {
490 select {
491 case <-stop:
492 return
493 default:
494 _ = a.Snapshot()
495 }
496 }
497 }()
498 wg.Wait()
499 a.Wait()
500 close(stop)
501 if a.Status() != StatusDone {
502 t.Fatalf("status = %s", a.Status())
503 }
504}
505
506func TestTruncateIsRuneAware(t *testing.T) {
507 input := "αβγδεζη"
508 if got := truncate(input, 6); got != "αβγ..." {
509 t.Fatalf("truncate(%q, 6) = %q; want %q", input, got, "αβγ...")
510 }
511 if got := truncate(input, 3); got != "..." {
512 t.Fatalf("truncate(%q, 3) = %q; want %q", input, got, "...")
513 }
514 if got := truncate(input, 2); got != ".." {
515 t.Fatalf("truncate(%q, 2) = %q; want %q", input, got, "..")
516 }
517}
518
519func TestTaskSlug(t *testing.T) {
520 cases := map[string]string{
521 "fix the login form": "fix-the-login-form",
522 " weird --- spaces!! ": "weird-spaces",
523 "": "agent",
524 "a-very-long-task-name-that-overflows": "a-very-long-task-name-th",
525 }
526 for in, want := range cases {
527 if got := taskSlug(in); got != want {
528 t.Errorf("taskSlug(%q) = %q; want %q", in, got, want)
529 }
530 }
531}
532
533func TestSnapshotAllSorted(t *testing.T) {
534 f := newTestSupervisor(t, func(a *Agent) Runner {
535 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
536 })
537 a1, _ := f.Spawn(context.Background(), "first")
538 // Force second spawn into a later nanosecond bucket.
539 time.Sleep(2 * time.Millisecond)
540 a2, _ := f.Spawn(context.Background(), "second")
541 a1.Wait()
542 a2.Wait()
543 snaps := f.SnapshotAll()
544 if len(snaps) != 2 {
545 t.Fatalf("want 2 snapshots; got %d", len(snaps))
546 }
547 if !snaps[0].Started.Before(snaps[1].Started) && !snaps[0].Started.Equal(snaps[1].Started) {
548 t.Fatal("snapshots not in spawn order")
549 }
550}