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 TestSpawnTimeoutDoesNotLimitResumableWorkerLifetime(t *testing.T) {
49 observedDeadline := make(chan bool, 1)
50 f := newTestSupervisor(t, func(a *Agent) Runner {
51 return RunnerFunc(func(ctx context.Context, _ Sink) error {
52 _, hasDeadline := ctx.Deadline()
53 observedDeadline <- hasDeadline
54 <-ctx.Done()
55 return ctx.Err()
56 })
57 })
58 a, err := f.SpawnReq(context.Background(), SpawnRequest{
59 Task: "remain resumable",
60 Timeout: time.Minute,
61 })
62 if err != nil {
63 t.Fatal(err)
64 }
65 if hasDeadline := <-observedDeadline; hasDeadline {
66 t.Fatal("worker lifetime context has a deadline; turn timeout must not expire an idle resumable worker")
67 }
68 if a.Timeout != time.Minute {
69 t.Fatalf("agent timeout = %s, want %s", a.Timeout, time.Minute)
70 }
71 f.StopAll()
72 a.Wait()
73}
74
75func TestSpawnRunsAndCompletes(t *testing.T) {
76 ran := make(chan string, 1)
77 f := newTestSupervisor(t, func(a *Agent) Runner {
78 return RunnerFunc(func(ctx context.Context, sink Sink) error {
79 ran <- a.Task
80 sink.Activity("hello")
81 sink.Transcript("line one")
82 sink.Transcript("line two")
83 return nil
84 })
85 })
86 a, err := f.Spawn(context.Background(), "do a thing")
87 if err != nil {
88 t.Fatal(err)
89 }
90 select {
91 case got := <-ran:
92 if got != "do a thing" {
93 t.Fatalf("runner got task %q; want %q", got, "do a thing")
94 }
95 case <-time.After(time.Second):
96 t.Fatal("runner did not start")
97 }
98 a.Wait()
99 if a.Status() != StatusDone {
100 t.Fatalf("status %s; want done", a.Status())
101 }
102 if got := a.Transcript(); len(got) != 2 || got[0] != "line one" || got[1] != "line two" {
103 t.Fatalf("transcript = %q", got)
104 }
105 if !strings.Contains(a.ID, "do-a-thing") {
106 t.Fatalf("id %q missing slug", a.ID)
107 }
108 // Every agent shares the host's RepoRoot.
109 if a.Dir != f.cfg.RepoRoot {
110 t.Fatalf("dir = %q; want repo root %q", a.Dir, f.cfg.RepoRoot)
111 }
112}
113
114// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
115// every spawned agent points its cwd at the parent zut's RepoRoot.
116func TestSpawnAgentSharesRepoRoot(t *testing.T) {
117 f := newTestSupervisor(t, func(a *Agent) Runner {
118 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
119 })
120 a, err := f.Spawn(context.Background(), "share me")
121 if err != nil {
122 t.Fatal(err)
123 }
124 a.Wait()
125 if a.Dir != f.cfg.RepoRoot {
126 t.Fatalf("Dir = %q; want RepoRoot %q", a.Dir, f.cfg.RepoRoot)
127 }
128}
129
130func TestSetRepoRootAffectsSubsequentSpawns(t *testing.T) {
131 f := newTestSupervisor(t, func(a *Agent) Runner {
132 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
133 })
134 want := t.TempDir()
135 f.SetRepoRoot(want)
136 a, err := f.Spawn(context.Background(), "use new root")
137 if err != nil {
138 t.Fatal(err)
139 }
140 a.Wait()
141 if a.Dir != want {
142 t.Fatalf("Dir = %q; want updated RepoRoot %q", a.Dir, want)
143 }
144}
145
146func TestSpawnEmptyTaskFails(t *testing.T) {
147 f := newTestSupervisor(t, func(a *Agent) Runner {
148 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
149 })
150 if _, err := f.Spawn(context.Background(), " "); err == nil {
151 t.Fatal("expected error on empty task")
152 }
153}
154
155func TestRunnerErrorMarksFailed(t *testing.T) {
156 wantErr := errors.New("boom")
157 f := newTestSupervisor(t, func(a *Agent) Runner {
158 return RunnerFunc(func(ctx context.Context, sink Sink) error { return wantErr })
159 })
160 a, _ := f.Spawn(context.Background(), "explode")
161 a.Wait()
162 if a.Status() != StatusFailed {
163 t.Fatalf("status %s; want failed", a.Status())
164 }
165 if !errors.Is(a.Err(), wantErr) {
166 t.Fatalf("err = %v", a.Err())
167 }
168 if !strings.Contains(a.Activity(), "boom") {
169 t.Fatalf("activity = %q; want it to mention the error", a.Activity())
170 }
171}
172
173func TestStopCancelsRunningAgent(t *testing.T) {
174 started := make(chan struct{})
175 f := newTestSupervisor(t, func(a *Agent) Runner {
176 return RunnerFunc(func(ctx context.Context, sink Sink) error {
177 close(started)
178 <-ctx.Done()
179 return ctx.Err()
180 })
181 })
182 a, _ := f.Spawn(context.Background(), "long")
183 <-started
184 if err := f.Stop(a.ID); err != nil {
185 t.Fatal(err)
186 }
187 a.Wait()
188 if a.Status() != StatusKilled {
189 t.Fatalf("status = %s; want killed", a.Status())
190 }
191}
192
193func TestStopByRootSessionLeavesOtherSessionsRunning(t *testing.T) {
194 started := make(chan string, 2)
195 stopped := make(chan string, 2)
196 f := newTestSupervisor(t, func(a *Agent) Runner {
197 return RunnerFunc(func(ctx context.Context, _ Sink) error {
198 started <- a.ID
199 <-ctx.Done()
200 stopped <- a.ID
201 return ctx.Err()
202 })
203 })
204 t.Cleanup(f.StopAll)
205
206 f.SetActiveSession("session-a")
207 a, err := f.Spawn(context.Background(), "session A task")
208 if err != nil {
209 t.Fatal(err)
210 }
211 f.SetActiveSession("session-b")
212 b, err := f.Spawn(context.Background(), "session B task")
213 if err != nil {
214 t.Fatal(err)
215 }
216
217 for range 2 {
218 <-started
219 }
220 if err := f.StopByRootSession("session-a"); err != nil {
221 t.Fatal(err)
222 }
223 if got := <-stopped; got != a.ID {
224 t.Fatalf("stopped agent = %q, want session A agent %q", got, a.ID)
225 }
226 select {
227 case got := <-stopped:
228 t.Fatalf("session cleanup also stopped %q", got)
229 default:
230 }
231 if got := b.Status(); got != StatusRunning {
232 t.Fatalf("session B status = %s, want running", got)
233 }
234}
235
236func TestSupervisorShutdownCommandsIdentifyOrigin(t *testing.T) {
237 if runtime.GOOS == "windows" {
238 t.Skip("subagent inbox transport uses Unix-domain sockets")
239 }
240 for _, tc := range []struct {
241 name string
242 origin ShutdownOrigin
243 stop func(*Supervisor, *Agent) error
244 }{
245 {name: "targeted", origin: ShutdownOriginTargeted, stop: func(f *Supervisor, a *Agent) error { return f.Stop(a.ID) }},
246 {name: "session", origin: ShutdownOriginSession, stop: func(f *Supervisor, _ *Agent) error { return f.StopByRootSession("session-a") }},
247 {name: "process", origin: ShutdownOriginProcess, stop: func(f *Supervisor, _ *Agent) error { return f.StopAllContext(context.Background()) }},
248 } {
249 t.Run(tc.name, func(t *testing.T) {
250 started := make(chan struct{})
251 commands := make(chan Envelope, 1)
252 f := newTestSupervisor(t, func(a *Agent) Runner {
253 return RunnerFunc(func(ctx context.Context, _ Sink) error {
254 listener, err := Listen(a.InboxPath)
255 if err != nil {
256 return err
257 }
258 defer listener.Close()
259 close(started)
260 select {
261 case line := <-listener.Lines():
262 command, err := ParseCommand(line)
263 if err != nil {
264 return err
265 }
266 commands <- command
267 return nil
268 case <-ctx.Done():
269 return ctx.Err()
270 }
271 })
272 })
273 f.SetActiveSession("session-a")
274 a, err := f.Spawn(context.Background(), "origin test")
275 if err != nil {
276 t.Fatal(err)
277 }
278 <-started
279 if err := tc.stop(f, a); err != nil {
280 t.Fatal(err)
281 }
282 command := <-commands
283 var payload AgentShutdownPayload
284 if err := command.DecodePayload(&payload); err != nil {
285 t.Fatal(err)
286 }
287 if payload.Origin != tc.origin {
288 t.Fatalf("shutdown origin = %q, want %q", payload.Origin, tc.origin)
289 }
290 a.Wait()
291 })
292 }
293}
294
295func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
296 if runtime.GOOS == "windows" {
297 t.Skip("subagent inbox transport uses Unix-domain sockets")
298 }
299 const grace = 250 * time.Millisecond
300 started := make(chan struct{})
301 canceled := make(chan struct{})
302 f := New(Config{
303 Root: t.TempDir(),
304 RepoRoot: t.TempDir(),
305 Policy: SubagentPolicy{
306 CancelGracePeriod: grace,
307 },
308 NewRunner: func(a *Agent) Runner {
309 return RunnerFunc(func(ctx context.Context, sink Sink) error {
310 listener, err := Listen(a.InboxPath)
311 if err != nil {
312 return err
313 }
314 defer listener.Close()
315 close(started)
316 select {
317 case <-listener.Lines():
318 case <-ctx.Done():
319 close(canceled)
320 return ctx.Err()
321 }
322 <-ctx.Done()
323 close(canceled)
324 return ctx.Err()
325 })
326 },
327 })
328 a, err := f.Spawn(context.Background(), "graceful stop")
329 if err != nil {
330 t.Fatal(err)
331 }
332 select {
333 case <-started:
334 case <-time.After(time.Second):
335 t.Fatal("runner did not start")
336 }
337
338 ctx, cancel := context.WithCancel(context.Background())
339 if err := f.StopContext(ctx, a.ID); err != nil {
340 t.Fatal(err)
341 }
342 cancel()
343
344 early := time.NewTimer(grace / 2)
345 defer early.Stop()
346 select {
347 case <-canceled:
348 t.Fatal("agent canceled when StopContext caller context was canceled")
349 case <-early.C:
350 }
351
352 deadline := time.NewTimer(grace * 2)
353 defer deadline.Stop()
354 select {
355 case <-canceled:
356 case <-deadline.C:
357 t.Fatal("agent was not canceled after the grace period")
358 }
359 a.Wait()
360}
361
362func TestStopContextCancelsDetachedWaitWithoutHoldingOperationLock(t *testing.T) {
363 if runtime.GOOS == "windows" {
364 t.Skip("subagent inbox transport uses Unix-domain sockets")
365 }
366 path := filepath.Join(shortSocketDir(t), "agent.sock")
367 listener, err := Listen(path)
368 if err != nil {
369 t.Fatal(err)
370 }
371 defer listener.Close()
372
373 f := New(Config{
374 Root: t.TempDir(),
375 Policy: SubagentPolicy{
376 CancelGracePeriod: time.Hour,
377 },
378 })
379 a := &Agent{
380 ID: "detached-agent",
381 InboxPath: path,
382 inbox: NewInbox(path),
383 status: StatusDetached,
384 done: make(chan struct{}),
385 }
386 defer a.inbox.Close()
387 close(a.done)
388 f.mu.Lock()
389 f.agents[a.ID] = a
390 f.order = append(f.order, a.ID)
391 f.mu.Unlock()
392
393 ctx, cancel := context.WithCancel(context.Background())
394 defer cancel()
395 stopDone := make(chan error, 1)
396 go func() { stopDone <- f.StopContext(ctx, a.ID) }()
397
398 // The listener notification proves Stop reached its shutdown wait. The
399 // operation lock must already be available while that wait is in progress.
400 if msg := <-listener.Lines(); msg == "" {
401 t.Fatal("shutdown command was empty")
402 }
403 operationReleased := make(chan struct{})
404 go func() {
405 f.operationMu.Lock()
406 close(operationReleased)
407 f.operationMu.Unlock()
408 }()
409 <-operationReleased
410
411 cancel()
412 if err := <-stopDone; !errors.Is(err, context.Canceled) {
413 t.Fatalf("StopContext error = %v; want context.Canceled", err)
414 }
415 if got := a.Status(); got != StatusDetached {
416 t.Fatalf("status after canceled stop = %s; want detached", got)
417 }
418}
419
420func TestStopDetachedWorkerClosesWaiters(t *testing.T) {
421 if runtime.GOOS == "windows" {
422 t.Skip("subagent inbox transport uses Unix-domain sockets")
423 }
424 path := filepath.Join(shortSocketDir(t), "agent.sock")
425 listener, err := Listen(path)
426 if err != nil {
427 t.Fatal(err)
428 }
429 t.Cleanup(func() { _ = listener.Close() })
430
431 root := t.TempDir()
432 f := New(Config{Root: root, RepoRoot: root})
433 stateDir := f.agentStateDir("detached-agent")
434 if err := os.MkdirAll(stateDir, 0o700); err != nil {
435 t.Fatal(err)
436 }
437 a := &Agent{
438 ID: "detached-agent",
439 stateDir: stateDir,
440 InboxPath: path,
441 inbox: NewInbox(path),
442 status: StatusDetached,
443 done: make(chan struct{}),
444 }
445 defer a.inbox.Close()
446 f.mu.Lock()
447 f.agents[a.ID] = a
448 f.order = append(f.order, a.ID)
449 f.mu.Unlock()
450
451 stopped := make(chan error, 1)
452 go func() { stopped <- f.StopContext(context.Background(), a.ID) }()
453 select {
454 case msg := <-listener.Lines():
455 if msg == "" {
456 t.Fatal("shutdown command was empty")
457 }
458 case <-time.After(time.Second):
459 t.Fatal("detached worker did not receive shutdown")
460 }
461 if err := listener.Close(); err != nil {
462 t.Fatal(err)
463 }
464 select {
465 case err := <-stopped:
466 if err != nil {
467 t.Fatalf("stop detached worker: %v", err)
468 }
469 case <-time.After(time.Second):
470 t.Fatal("detached worker stop did not finish")
471 }
472 select {
473 case <-a.done:
474 case <-time.After(time.Second):
475 t.Fatal("detached worker stop did not close waiters")
476 }
477 result, err := readTurnResult(stateDir)
478 if err != nil {
479 t.Fatal(err)
480 }
481 if result.AgentID != a.ID || result.Status != ResultCanceled {
482 t.Fatalf("detached worker result = agent %q status %q, want agent %q status %q", result.AgentID, result.Status, a.ID, ResultCanceled)
483 }
484}
485
486func TestStopDetachedWorkerHonorsCanceledShutdownSend(t *testing.T) {
487 if runtime.GOOS == "windows" {
488 t.Skip("subagent inbox transport uses Unix-domain sockets")
489 }
490 path := filepath.Join(shortSocketDir(t), "agent.sock")
491 listener, err := Listen(path)
492 if err != nil {
493 t.Fatal(err)
494 }
495 defer listener.Close()
496
497 local, remote := net.Pipe()
498 defer remote.Close()
499 conn := &blockingWriteConn{
500 Conn: local,
501 started: make(chan struct{}),
502 release: make(chan struct{}),
503 }
504 inbox := NewInbox(path)
505 inbox.conn = conn
506 defer inbox.Close()
507
508 f := New(Config{Root: t.TempDir()})
509 a := &Agent{
510 ID: "detached-cancel",
511 InboxPath: path,
512 inbox: inbox,
513 status: StatusDetached,
514 done: make(chan struct{}),
515 }
516 f.mu.Lock()
517 f.agents[a.ID] = a
518 f.order = append(f.order, a.ID)
519 f.mu.Unlock()
520
521 ctx, cancel := context.WithCancel(context.Background())
522 stopped := make(chan error, 1)
523 go func() { stopped <- f.StopContext(ctx, a.ID) }()
524 select {
525 case <-conn.started:
526 case <-time.After(time.Second):
527 t.Fatal("detached worker shutdown send did not start")
528 }
529 cancel()
530 select {
531 case err := <-stopped:
532 if !errors.Is(err, context.Canceled) {
533 t.Fatalf("stop detached worker error = %v, want context canceled", err)
534 }
535 case <-time.After(time.Second):
536 t.Fatal("detached worker shutdown send did not honor cancellation")
537 }
538}
539
540func TestStopAfterDoneIsNoop(t *testing.T) {
541 f := newTestSupervisor(t, func(a *Agent) Runner {
542 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
543 })
544 a, _ := f.Spawn(context.Background(), "quick")
545 a.Wait()
546 if err := f.Stop(a.ID); err != nil {
547 t.Fatalf("stop after done: %v", err)
548 }
549 if a.Status() != StatusDone {
550 t.Fatalf("status flipped to %s", a.Status())
551 }
552}
553
554func TestGetPrefixMatch(t *testing.T) {
555 f := newTestSupervisor(t, func(a *Agent) Runner {
556 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
557 })
558 a, _ := f.Spawn(context.Background(), "alpha task")
559 a.Wait()
560 // Full id works.
561 if got := f.Get(a.ID); got != a {
562 t.Fatal("get by full id failed")
563 }
564 // Slug prefix works as long as it's unique.
565 if got := f.Get("alpha"); got != a {
566 t.Fatal("get by prefix failed")
567 }
568 // Bogus id returns nil.
569 if got := f.Get("zzz-nope"); got != nil {
570 t.Fatalf("expected nil; got %#v", got)
571 }
572}
573
574func TestRemoveRequiresTerminalState(t *testing.T) {
575 hold := make(chan struct{})
576 f := newTestSupervisor(t, func(a *Agent) Runner {
577 return RunnerFunc(func(ctx context.Context, sink Sink) error {
578 <-hold
579 return nil
580 })
581 })
582 a, _ := f.Spawn(context.Background(), "still going")
583 // Wait for run goroutine to flip to running.
584 for i := 0; i < 100 && a.Status() != StatusRunning; i++ {
585 time.Sleep(time.Millisecond)
586 }
587 if err := f.Remove(a.ID); err == nil {
588 t.Fatal("remove of running agent should fail")
589 }
590 close(hold)
591 a.Wait()
592 if err := f.Remove(a.ID); err != nil {
593 t.Fatalf("remove after done: %v", err)
594 }
595 if got := f.Get(a.ID); got != nil {
596 t.Fatal("agent still present after remove")
597 }
598}
599
600func TestSnapshotIsStableAcrossAccess(t *testing.T) {
601 var wg sync.WaitGroup
602 wg.Add(1)
603 f := newTestSupervisor(t, func(a *Agent) Runner {
604 return RunnerFunc(func(ctx context.Context, sink Sink) error {
605 defer wg.Done()
606 for i := 0; i < 200; i++ {
607 sink.Transcript("chunk")
608 sink.Activity("step")
609 }
610 return nil
611 })
612 })
613 a, _ := f.Spawn(context.Background(), "race")
614 // Hammer Snapshot while the runner is writing; the -race detector
615 // is the real assertion here.
616 stop := make(chan struct{})
617 go func() {
618 for {
619 select {
620 case <-stop:
621 return
622 default:
623 _ = a.Snapshot()
624 }
625 }
626 }()
627 wg.Wait()
628 a.Wait()
629 close(stop)
630 if a.Status() != StatusDone {
631 t.Fatalf("status = %s", a.Status())
632 }
633}
634
635func TestTruncateIsRuneAware(t *testing.T) {
636 input := "αβγδεζη"
637 if got := truncate(input, 6); got != "αβγ..." {
638 t.Fatalf("truncate(%q, 6) = %q; want %q", input, got, "αβγ...")
639 }
640 if got := truncate(input, 3); got != "..." {
641 t.Fatalf("truncate(%q, 3) = %q; want %q", input, got, "...")
642 }
643 if got := truncate(input, 2); got != ".." {
644 t.Fatalf("truncate(%q, 2) = %q; want %q", input, got, "..")
645 }
646}
647
648func TestTaskSlug(t *testing.T) {
649 cases := map[string]string{
650 "fix the login form": "fix-the-login-form",
651 " weird --- spaces!! ": "weird-spaces",
652 "": "agent",
653 "a-very-long-task-name-that-overflows": "a-very-long-task-name-th",
654 }
655 for in, want := range cases {
656 if got := taskSlug(in); got != want {
657 t.Errorf("taskSlug(%q) = %q; want %q", in, got, want)
658 }
659 }
660}
661
662func TestSnapshotAllSorted(t *testing.T) {
663 f := newTestSupervisor(t, func(a *Agent) Runner {
664 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
665 })
666 a1, _ := f.Spawn(context.Background(), "first")
667 // Force second spawn into a later nanosecond bucket.
668 time.Sleep(2 * time.Millisecond)
669 a2, _ := f.Spawn(context.Background(), "second")
670 a1.Wait()
671 a2.Wait()
672 snaps := f.SnapshotAll()
673 if len(snaps) != 2 {
674 t.Fatalf("want 2 snapshots; got %d", len(snaps))
675 }
676 if !snaps[0].Started.Before(snaps[1].Started) && !snaps[0].Started.Equal(snaps[1].Started) {
677 t.Fatal("snapshots not in spawn order")
678 }
679}