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
114func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
115 root := t.TempDir()
116 started := make(chan struct{})
117 cancelled := make(chan struct{})
118 startupTimeout := time.Hour
119 f := New(Config{
120 Root: root, RepoRoot: root,
121 Policy: SubagentPolicy{StartupTimeout: startupTimeout},
122 NewRunner: func(*Agent) Runner {
123 return RunnerFunc(func(ctx context.Context, _ Sink) error {
124 close(started)
125 <-ctx.Done()
126 close(cancelled)
127 return ctx.Err()
128 })
129 },
130 })
131 t.Cleanup(f.StopAll)
132
133 a, err := f.Spawn(context.Background(), "stuck startup")
134 if err != nil {
135 t.Fatal(err)
136 }
137 select {
138 case <-started:
139 case <-time.After(time.Second):
140 t.Fatal("runner did not start")
141 }
142 f.expireStarting(a, startupTimeout)
143 select {
144 case <-cancelled:
145 case <-time.After(time.Second):
146 t.Fatal("startup timeout did not cancel runner")
147 }
148 waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
149 defer cancel()
150 if err := a.WaitContext(waitCtx); err != nil {
151 t.Fatalf("worker did not fail after startup timeout: %v", err)
152 }
153 if got := a.Status(); got != StatusFailed {
154 t.Fatalf("status = %s, want %s", got, StatusFailed)
155 }
156 if !errors.Is(a.Err(), errStartupTimeout) {
157 t.Fatalf("worker error = %v, want startup timeout", a.Err())
158 }
159 if got := a.Activity(); !strings.Contains(got, "startup timeout") {
160 t.Fatalf("activity = %q, want startup timeout explanation", got)
161 }
162 result, err := f.ReadResult(a.ID)
163 if err != nil {
164 t.Fatal(err)
165 }
166 if result.Status != ResultFailed || result.Error == nil {
167 t.Fatalf("startup timeout result = %#v, want failed result with error", result)
168 }
169 if !strings.Contains(result.Error.Message, errStartupTimeout.Error()) || !strings.Contains(result.Error.Message, "inspect the worker output and retry") {
170 t.Fatalf("startup timeout result error = %#v, want timeout and retry guidance", result.Error)
171 }
172}
173
174func TestStartupTimeoutDoesNotCancelReadyWorker(t *testing.T) {
175 root := t.TempDir()
176 started := make(chan struct{})
177 cancelled := make(chan struct{})
178 f := New(Config{
179 Root: root, RepoRoot: root,
180 Policy: SubagentPolicy{StartupTimeout: time.Hour},
181 NewRunner: func(*Agent) Runner {
182 return RunnerFunc(func(ctx context.Context, _ Sink) error {
183 close(started)
184 <-ctx.Done()
185 close(cancelled)
186 return ctx.Err()
187 })
188 },
189 })
190 t.Cleanup(f.StopAll)
191
192 a, err := f.Spawn(context.Background(), "ready startup")
193 if err != nil {
194 t.Fatal(err)
195 }
196 select {
197 case <-started:
198 case <-time.After(time.Second):
199 t.Fatal("runner did not start")
200 }
201 if err := updateAgentFromEvent(a, NewEvent(EventAgentReady, nil)); err != nil {
202 t.Fatal(err)
203 }
204 f.expireStarting(a, time.Hour)
205
206 if got := a.ProcessState(); got != ProcessAlive {
207 t.Fatalf("process state = %s, want %s", got, ProcessAlive)
208 }
209 if err := a.Err(); err != nil {
210 t.Fatalf("ready worker error = %v, want nil", err)
211 }
212 select {
213 case <-cancelled:
214 t.Fatal("startup timeout canceled ready worker")
215 default:
216 }
217 if err := f.Stop(a.ID); err != nil {
218 t.Fatal(err)
219 }
220 a.Wait()
221}
222
223func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
224 root := t.TempDir()
225 started := make(chan struct{})
226 queueTimeout := time.Hour
227 var startOnce sync.Once
228 f := New(Config{
229 Root: root, RepoRoot: root,
230 Policy: SubagentPolicy{
231 MaxConcurrent: 1,
232 QueueTimeout: queueTimeout,
233 StartupTimeout: time.Hour,
234 },
235 NewRunner: func(*Agent) Runner {
236 return RunnerFunc(func(ctx context.Context, _ Sink) error {
237 startOnce.Do(func() { close(started) })
238 <-ctx.Done()
239 return ctx.Err()
240 })
241 },
242 })
243 t.Cleanup(f.StopAll)
244
245 first, err := f.Spawn(context.Background(), "occupy only slot")
246 if err != nil {
247 t.Fatal(err)
248 }
249 select {
250 case <-started:
251 case <-time.After(time.Second):
252 t.Fatal("first runner did not start")
253 }
254 queued, err := f.Spawn(context.Background(), "wait for slot")
255 if err != nil {
256 t.Fatal(err)
257 }
258 f.expireQueued(queued.ID)
259 waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
260 defer cancel()
261 if err := queued.WaitContext(waitCtx); err != nil {
262 t.Fatalf("queued worker did not time out: %v", err)
263 }
264 if got := queued.Status(); got != StatusFailed {
265 t.Fatalf("queued worker status = %s, want %s", got, StatusFailed)
266 }
267 if got := queued.ProcessState(); got != ProcessExited {
268 t.Fatalf("queued worker process state = %s, want %s", got, ProcessExited)
269 }
270 if got := queued.Err(); got == nil || !strings.Contains(got.Error(), "waiting for an execution slot") {
271 t.Fatalf("queued worker error = %v, want queue-pressure explanation", got)
272 }
273 result, err := f.ReadResult(queued.ID)
274 if err != nil {
275 t.Fatal(err)
276 }
277 if result.Status != ResultFailed || result.Error == nil || !strings.Contains(result.Error.Message, "waiting for an execution slot") {
278 t.Fatalf("queued worker result = %#v, want failed result with queue-pressure error", result)
279 }
280 if err := f.Stop(first.ID); err != nil {
281 t.Fatal(err)
282 }
283 first.Wait()
284}
285
286// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
287// every spawned agent points its cwd at the parent zut's RepoRoot.
288func TestSpawnAgentSharesRepoRoot(t *testing.T) {
289 f := newTestSupervisor(t, func(a *Agent) Runner {
290 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
291 })
292 a, err := f.Spawn(context.Background(), "share me")
293 if err != nil {
294 t.Fatal(err)
295 }
296 a.Wait()
297 if a.Dir != f.cfg.RepoRoot {
298 t.Fatalf("Dir = %q; want RepoRoot %q", a.Dir, f.cfg.RepoRoot)
299 }
300}
301
302func TestSetRepoRootAffectsSubsequentSpawns(t *testing.T) {
303 f := newTestSupervisor(t, func(a *Agent) Runner {
304 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
305 })
306 want := t.TempDir()
307 f.SetRepoRoot(want)
308 a, err := f.Spawn(context.Background(), "use new root")
309 if err != nil {
310 t.Fatal(err)
311 }
312 a.Wait()
313 if a.Dir != want {
314 t.Fatalf("Dir = %q; want updated RepoRoot %q", a.Dir, want)
315 }
316}
317
318func TestSpawnEmptyTaskFails(t *testing.T) {
319 f := newTestSupervisor(t, func(a *Agent) Runner {
320 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
321 })
322 if _, err := f.Spawn(context.Background(), " "); err == nil {
323 t.Fatal("expected error on empty task")
324 }
325}
326
327func TestRunnerErrorMarksFailed(t *testing.T) {
328 wantErr := errors.New("boom")
329 f := newTestSupervisor(t, func(a *Agent) Runner {
330 return RunnerFunc(func(ctx context.Context, sink Sink) error { return wantErr })
331 })
332 a, _ := f.Spawn(context.Background(), "explode")
333 a.Wait()
334 if a.Status() != StatusFailed {
335 t.Fatalf("status %s; want failed", a.Status())
336 }
337 if !errors.Is(a.Err(), wantErr) {
338 t.Fatalf("err = %v", a.Err())
339 }
340 if !strings.Contains(a.Activity(), "boom") {
341 t.Fatalf("activity = %q; want it to mention the error", a.Activity())
342 }
343}
344
345func TestStopCancelsRunningAgent(t *testing.T) {
346 started := make(chan struct{})
347 f := newTestSupervisor(t, func(a *Agent) Runner {
348 return RunnerFunc(func(ctx context.Context, sink Sink) error {
349 close(started)
350 <-ctx.Done()
351 return ctx.Err()
352 })
353 })
354 a, _ := f.Spawn(context.Background(), "long")
355 <-started
356 if err := f.Stop(a.ID); err != nil {
357 t.Fatal(err)
358 }
359 a.Wait()
360 if a.Status() != StatusKilled {
361 t.Fatalf("status = %s; want killed", a.Status())
362 }
363}
364
365func TestStopPublishesShutdownOriginBeforeConcurrentFinalization(t *testing.T) {
366 started := make(chan struct{})
367 finish := make(chan struct{})
368 finalizing := make(chan struct{})
369 f := newTestSupervisor(t, func(a *Agent) Runner {
370 return RunnerFunc(func(context.Context, Sink) error {
371 close(started)
372 <-finish
373 close(finalizing)
374 return nil
375 })
376 })
377 a, err := f.Spawn(context.Background(), "finish while stopping")
378 if err != nil {
379 t.Fatal(err)
380 }
381 <-started
382
383 // Hold lifecycle finalization after Runner.Run returns. Shutdown must publish
384 // its origin before exposing StatusKilled, without relying on finalization
385 // ordering to fill the durable result correctly.
386 a.lifecycleMu.Lock()
387 lifecycleLocked := true
388 finishClosed := false
389 defer func() {
390 if !finishClosed {
391 close(finish)
392 }
393 if lifecycleLocked {
394 a.lifecycleMu.Unlock()
395 }
396 }()
397 stopDone := make(chan error, 1)
398 go func() { stopDone <- f.stop(context.Background(), a.ID, ShutdownOriginSession) }()
399
400 deadline := time.Now().Add(time.Second)
401 for a.Status() != StatusKilled {
402 if time.Now().After(deadline) {
403 t.Fatal("shutdown did not publish killed status")
404 }
405 runtime.Gosched()
406 }
407 a.mu.Lock()
408 origin := a.shutdownOrigin
409 a.mu.Unlock()
410 if origin != ShutdownOriginSession {
411 t.Fatalf("shutdown origin = %q when killed became observable, want %q", origin, ShutdownOriginSession)
412 }
413
414 close(finish)
415 finishClosed = true
416 <-finalizing
417 deadline = time.Now().Add(time.Second)
418 for {
419 a.mu.Lock()
420 finalizationStarted := !a.finished.IsZero()
421 a.mu.Unlock()
422 if finalizationStarted {
423 break
424 }
425 if time.Now().After(deadline) {
426 t.Fatal("runner finalization did not start")
427 }
428 runtime.Gosched()
429 }
430 a.lifecycleMu.Unlock()
431 lifecycleLocked = false
432 if err := <-stopDone; err != nil {
433 t.Fatal(err)
434 }
435 a.Wait()
436
437 result := a.Result()
438 if result == nil || result.Status != ResultCanceled || result.ShutdownOrigin != ShutdownOriginSession {
439 t.Fatalf("final result = %#v, want attributed session cancellation", result)
440 }
441 if result.Error == nil || result.Error.Code != "shutdown" {
442 t.Fatalf("final result error = %#v, want shutdown", result.Error)
443 }
444}
445
446func TestStopByRootSessionLeavesOtherSessionsRunning(t *testing.T) {
447 started := make(chan string, 2)
448 stopped := make(chan string, 2)
449 f := newTestSupervisor(t, func(a *Agent) Runner {
450 return RunnerFunc(func(ctx context.Context, _ Sink) error {
451 started <- a.ID
452 <-ctx.Done()
453 stopped <- a.ID
454 return ctx.Err()
455 })
456 })
457 t.Cleanup(f.StopAll)
458
459 f.SetActiveSession("session-a")
460 a, err := f.Spawn(context.Background(), "session A task")
461 if err != nil {
462 t.Fatal(err)
463 }
464 f.SetActiveSession("session-b")
465 b, err := f.Spawn(context.Background(), "session B task")
466 if err != nil {
467 t.Fatal(err)
468 }
469
470 for range 2 {
471 <-started
472 }
473 if err := f.StopByRootSession("session-a"); err != nil {
474 t.Fatal(err)
475 }
476 if got := <-stopped; got != a.ID {
477 t.Fatalf("stopped agent = %q, want session A agent %q", got, a.ID)
478 }
479 select {
480 case got := <-stopped:
481 t.Fatalf("session cleanup also stopped %q", got)
482 default:
483 }
484 if got := b.Status(); got != StatusRunning {
485 t.Fatalf("session B status = %s, want running", got)
486 }
487}
488
489func TestSupervisorShutdownCommandsIdentifyOrigin(t *testing.T) {
490 if runtime.GOOS == "windows" {
491 t.Skip("subagent inbox transport uses Unix-domain sockets")
492 }
493 for _, tc := range []struct {
494 name string
495 origin ShutdownOrigin
496 stop func(*Supervisor, *Agent) error
497 }{
498 {name: "targeted", origin: ShutdownOriginTargeted, stop: func(f *Supervisor, a *Agent) error { return f.Stop(a.ID) }},
499 {name: "session", origin: ShutdownOriginSession, stop: func(f *Supervisor, _ *Agent) error { return f.StopByRootSession("session-a") }},
500 {name: "process", origin: ShutdownOriginProcess, stop: func(f *Supervisor, _ *Agent) error { return f.StopAllContext(context.Background()) }},
501 } {
502 t.Run(tc.name, func(t *testing.T) {
503 started := make(chan struct{})
504 commands := make(chan Envelope, 1)
505 f := newTestSupervisor(t, func(a *Agent) Runner {
506 return RunnerFunc(func(ctx context.Context, _ Sink) error {
507 listener, err := Listen(a.InboxPath)
508 if err != nil {
509 return err
510 }
511 defer listener.Close()
512 close(started)
513 select {
514 case line := <-listener.Lines():
515 command, err := ParseCommand(line)
516 if err != nil {
517 return err
518 }
519 commands <- command
520 return nil
521 case <-ctx.Done():
522 return ctx.Err()
523 }
524 })
525 })
526 f.SetActiveSession("session-a")
527 a, err := f.Spawn(context.Background(), "origin test")
528 if err != nil {
529 t.Fatal(err)
530 }
531 <-started
532 if err := tc.stop(f, a); err != nil {
533 t.Fatal(err)
534 }
535 command := <-commands
536 var payload AgentShutdownPayload
537 if err := command.DecodePayload(&payload); err != nil {
538 t.Fatal(err)
539 }
540 if payload.Origin != tc.origin {
541 t.Fatalf("shutdown origin = %q, want %q", payload.Origin, tc.origin)
542 }
543 a.Wait()
544 })
545 }
546}
547
548func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
549 if runtime.GOOS == "windows" {
550 t.Skip("subagent inbox transport uses Unix-domain sockets")
551 }
552 const grace = 250 * time.Millisecond
553 started := make(chan struct{})
554 canceled := make(chan struct{})
555 f := New(Config{
556 Root: t.TempDir(),
557 RepoRoot: t.TempDir(),
558 Policy: SubagentPolicy{
559 CancelGracePeriod: grace,
560 },
561 NewRunner: func(a *Agent) Runner {
562 return RunnerFunc(func(ctx context.Context, sink Sink) error {
563 listener, err := Listen(a.InboxPath)
564 if err != nil {
565 return err
566 }
567 defer listener.Close()
568 close(started)
569 select {
570 case <-listener.Lines():
571 case <-ctx.Done():
572 close(canceled)
573 return ctx.Err()
574 }
575 <-ctx.Done()
576 close(canceled)
577 return ctx.Err()
578 })
579 },
580 })
581 a, err := f.Spawn(context.Background(), "graceful stop")
582 if err != nil {
583 t.Fatal(err)
584 }
585 select {
586 case <-started:
587 case <-time.After(time.Second):
588 t.Fatal("runner did not start")
589 }
590
591 ctx, cancel := context.WithCancel(context.Background())
592 if err := f.StopContext(ctx, a.ID); err != nil {
593 t.Fatal(err)
594 }
595 cancel()
596
597 early := time.NewTimer(grace / 2)
598 defer early.Stop()
599 select {
600 case <-canceled:
601 t.Fatal("agent canceled when StopContext caller context was canceled")
602 case <-early.C:
603 }
604
605 deadline := time.NewTimer(grace * 2)
606 defer deadline.Stop()
607 select {
608 case <-canceled:
609 case <-deadline.C:
610 t.Fatal("agent was not canceled after the grace period")
611 }
612 a.Wait()
613}
614
615func TestStopContextCancelsDetachedWaitWithoutHoldingOperationLock(t *testing.T) {
616 if runtime.GOOS == "windows" {
617 t.Skip("subagent inbox transport uses Unix-domain sockets")
618 }
619 path := filepath.Join(shortSocketDir(t), "agent.sock")
620 listener, err := Listen(path)
621 if err != nil {
622 t.Fatal(err)
623 }
624 defer listener.Close()
625
626 f := New(Config{
627 Root: t.TempDir(),
628 Policy: SubagentPolicy{
629 CancelGracePeriod: time.Hour,
630 },
631 })
632 a := &Agent{
633 ID: "detached-agent",
634 InboxPath: path,
635 inbox: NewInbox(path),
636 status: StatusDetached,
637 done: make(chan struct{}),
638 }
639 defer a.inbox.Close()
640 close(a.done)
641 f.mu.Lock()
642 f.agents[a.ID] = a
643 f.order = append(f.order, a.ID)
644 f.mu.Unlock()
645
646 ctx, cancel := context.WithCancel(context.Background())
647 defer cancel()
648 stopDone := make(chan error, 1)
649 go func() { stopDone <- f.StopContext(ctx, a.ID) }()
650
651 // The listener notification proves Stop reached its shutdown wait. The
652 // operation lock must already be available while that wait is in progress.
653 if msg := <-listener.Lines(); msg == "" {
654 t.Fatal("shutdown command was empty")
655 }
656 operationReleased := make(chan struct{})
657 go func() {
658 f.operationMu.Lock()
659 close(operationReleased)
660 f.operationMu.Unlock()
661 }()
662 <-operationReleased
663
664 cancel()
665 if err := <-stopDone; !errors.Is(err, context.Canceled) {
666 t.Fatalf("StopContext error = %v; want context.Canceled", err)
667 }
668 if got := a.Status(); got != StatusDetached {
669 t.Fatalf("status after canceled stop = %s; want detached", got)
670 }
671}
672
673func TestStopDetachedWorkerClosesWaiters(t *testing.T) {
674 if runtime.GOOS == "windows" {
675 t.Skip("subagent inbox transport uses Unix-domain sockets")
676 }
677 path := filepath.Join(shortSocketDir(t), "agent.sock")
678 listener, err := Listen(path)
679 if err != nil {
680 t.Fatal(err)
681 }
682 t.Cleanup(func() { _ = listener.Close() })
683
684 root := t.TempDir()
685 f := New(Config{Root: root, RepoRoot: root})
686 stateDir := f.agentStateDir("detached-agent")
687 if err := os.MkdirAll(stateDir, 0o700); err != nil {
688 t.Fatal(err)
689 }
690 a := &Agent{
691 ID: "detached-agent",
692 stateDir: stateDir,
693 InboxPath: path,
694 inbox: NewInbox(path),
695 status: StatusDetached,
696 done: make(chan struct{}),
697 }
698 defer a.inbox.Close()
699 f.mu.Lock()
700 f.agents[a.ID] = a
701 f.order = append(f.order, a.ID)
702 f.mu.Unlock()
703
704 stopped := make(chan error, 1)
705 go func() { stopped <- f.StopContext(context.Background(), a.ID) }()
706 select {
707 case msg := <-listener.Lines():
708 if msg == "" {
709 t.Fatal("shutdown command was empty")
710 }
711 case <-time.After(time.Second):
712 t.Fatal("detached worker did not receive shutdown")
713 }
714 if err := listener.Close(); err != nil {
715 t.Fatal(err)
716 }
717 select {
718 case err := <-stopped:
719 if err != nil {
720 t.Fatalf("stop detached worker: %v", err)
721 }
722 case <-time.After(time.Second):
723 t.Fatal("detached worker stop did not finish")
724 }
725 select {
726 case <-a.done:
727 case <-time.After(time.Second):
728 t.Fatal("detached worker stop did not close waiters")
729 }
730 result, err := readTurnResult(stateDir)
731 if err != nil {
732 t.Fatal(err)
733 }
734 if result.AgentID != a.ID || result.Status != ResultCanceled {
735 t.Fatalf("detached worker result = agent %q status %q, want agent %q status %q", result.AgentID, result.Status, a.ID, ResultCanceled)
736 }
737}
738
739func TestStopDetachedWorkerHonorsCanceledShutdownSend(t *testing.T) {
740 if runtime.GOOS == "windows" {
741 t.Skip("subagent inbox transport uses Unix-domain sockets")
742 }
743 path := filepath.Join(shortSocketDir(t), "agent.sock")
744 listener, err := Listen(path)
745 if err != nil {
746 t.Fatal(err)
747 }
748 defer listener.Close()
749
750 local, remote := net.Pipe()
751 defer remote.Close()
752 conn := &blockingWriteConn{
753 Conn: local,
754 started: make(chan struct{}),
755 release: make(chan struct{}),
756 }
757 inbox := NewInbox(path)
758 inbox.conn = conn
759 defer inbox.Close()
760
761 f := New(Config{Root: t.TempDir()})
762 a := &Agent{
763 ID: "detached-cancel",
764 InboxPath: path,
765 inbox: inbox,
766 status: StatusDetached,
767 done: make(chan struct{}),
768 }
769 f.mu.Lock()
770 f.agents[a.ID] = a
771 f.order = append(f.order, a.ID)
772 f.mu.Unlock()
773
774 ctx, cancel := context.WithCancel(context.Background())
775 stopped := make(chan error, 1)
776 go func() { stopped <- f.StopContext(ctx, a.ID) }()
777 select {
778 case <-conn.started:
779 case <-time.After(time.Second):
780 t.Fatal("detached worker shutdown send did not start")
781 }
782 cancel()
783 select {
784 case err := <-stopped:
785 if !errors.Is(err, context.Canceled) {
786 t.Fatalf("stop detached worker error = %v, want context canceled", err)
787 }
788 case <-time.After(time.Second):
789 t.Fatal("detached worker shutdown send did not honor cancellation")
790 }
791}
792
793func TestStopAfterDoneIsNoop(t *testing.T) {
794 f := newTestSupervisor(t, func(a *Agent) Runner {
795 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
796 })
797 a, _ := f.Spawn(context.Background(), "quick")
798 a.Wait()
799 if err := f.Stop(a.ID); err != nil {
800 t.Fatalf("stop after done: %v", err)
801 }
802 if a.Status() != StatusDone {
803 t.Fatalf("status flipped to %s", a.Status())
804 }
805}
806
807func TestGetPrefixMatch(t *testing.T) {
808 f := newTestSupervisor(t, func(a *Agent) Runner {
809 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
810 })
811 a, _ := f.Spawn(context.Background(), "alpha task")
812 a.Wait()
813 // Full id works.
814 if got := f.Get(a.ID); got != a {
815 t.Fatal("get by full id failed")
816 }
817 // Slug prefix works as long as it's unique.
818 if got := f.Get("alpha"); got != a {
819 t.Fatal("get by prefix failed")
820 }
821 // Bogus id returns nil.
822 if got := f.Get("zzz-nope"); got != nil {
823 t.Fatalf("expected nil; got %#v", got)
824 }
825}
826
827func TestRemoveRequiresTerminalState(t *testing.T) {
828 hold := make(chan struct{})
829 f := newTestSupervisor(t, func(a *Agent) Runner {
830 return RunnerFunc(func(ctx context.Context, sink Sink) error {
831 <-hold
832 return nil
833 })
834 })
835 a, _ := f.Spawn(context.Background(), "still going")
836 // Wait for run goroutine to flip to running.
837 for i := 0; i < 100 && a.Status() != StatusRunning; i++ {
838 time.Sleep(time.Millisecond)
839 }
840 if err := f.Remove(a.ID); err == nil {
841 t.Fatal("remove of running agent should fail")
842 }
843 close(hold)
844 a.Wait()
845 if err := f.Remove(a.ID); err != nil {
846 t.Fatalf("remove after done: %v", err)
847 }
848 if got := f.Get(a.ID); got != nil {
849 t.Fatal("agent still present after remove")
850 }
851}
852
853func TestSnapshotIsStableAcrossAccess(t *testing.T) {
854 var wg sync.WaitGroup
855 wg.Add(1)
856 f := newTestSupervisor(t, func(a *Agent) Runner {
857 return RunnerFunc(func(ctx context.Context, sink Sink) error {
858 defer wg.Done()
859 for i := 0; i < 200; i++ {
860 sink.Transcript("chunk")
861 sink.Activity("step")
862 }
863 return nil
864 })
865 })
866 a, _ := f.Spawn(context.Background(), "race")
867 // Hammer Snapshot while the runner is writing; the -race detector
868 // is the real assertion here.
869 stop := make(chan struct{})
870 go func() {
871 for {
872 select {
873 case <-stop:
874 return
875 default:
876 _ = a.Snapshot()
877 }
878 }
879 }()
880 wg.Wait()
881 a.Wait()
882 close(stop)
883 if a.Status() != StatusDone {
884 t.Fatalf("status = %s", a.Status())
885 }
886}
887
888func TestTruncateIsRuneAware(t *testing.T) {
889 input := "αβγδεζη"
890 if got := truncate(input, 6); got != "αβγ..." {
891 t.Fatalf("truncate(%q, 6) = %q; want %q", input, got, "αβγ...")
892 }
893 if got := truncate(input, 3); got != "..." {
894 t.Fatalf("truncate(%q, 3) = %q; want %q", input, got, "...")
895 }
896 if got := truncate(input, 2); got != ".." {
897 t.Fatalf("truncate(%q, 2) = %q; want %q", input, got, "..")
898 }
899}
900
901func TestTaskSlug(t *testing.T) {
902 cases := map[string]string{
903 "fix the login form": "fix-the-login-form",
904 " weird --- spaces!! ": "weird-spaces",
905 "": "agent",
906 "a-very-long-task-name-that-overflows": "a-very-long-task-name-th",
907 }
908 for in, want := range cases {
909 if got := taskSlug(in); got != want {
910 t.Errorf("taskSlug(%q) = %q; want %q", in, got, want)
911 }
912 }
913}
914
915func TestSnapshotAllSorted(t *testing.T) {
916 f := newTestSupervisor(t, func(a *Agent) Runner {
917 return RunnerFunc(func(ctx context.Context, sink Sink) error { return nil })
918 })
919 a1, _ := f.Spawn(context.Background(), "first")
920 // Force second spawn into a later nanosecond bucket.
921 time.Sleep(2 * time.Millisecond)
922 a2, _ := f.Spawn(context.Background(), "second")
923 a1.Wait()
924 a2.Wait()
925 snaps := f.SnapshotAll()
926 if len(snaps) != 2 {
927 t.Fatalf("want 2 snapshots; got %d", len(snaps))
928 }
929 if !snaps[0].Started.Before(snaps[1].Started) && !snaps[0].Started.Equal(snaps[1].Started) {
930 t.Fatal("snapshots not in spawn order")
931 }
932}