1package subagents
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "os"
8 "path/filepath"
9 "runtime"
10 "strings"
11 "sync"
12 "sync/atomic"
13 "testing"
14 "time"
15)
16
17// TestSpawnWritesMetaJSON asserts the durability contract: every
18// successful Spawn leaves a meta.json on disk with the agent's
19// identity bits. Without this, Reload on the next zut launch can't
20// find the agent and the user loses access to the worktree.
21func TestSpawnWritesMetaJSON(t *testing.T) {
22 root := t.TempDir()
23 f := New(Config{
24 Root: root,
25 RepoRoot: root,
26 NewRunner: func(a *Agent) Runner {
27 return RunnerFunc(func(ctx context.Context, _ Sink) error {
28 <-ctx.Done()
29 return ctx.Err()
30 })
31 },
32 })
33 defer f.StopAll()
34
35 a, err := f.Spawn(context.Background(), "investigate widget")
36 if err != nil {
37 t.Fatalf("spawn: %v", err)
38 }
39
40 // Read meta.json straight off disk.
41 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
42 if err != nil {
43 t.Fatalf("read meta.json: %v", err)
44 }
45 var got agentMeta
46 if err := json.Unmarshal(metaBytes, &got); err != nil {
47 t.Fatalf("parse meta.json: %v", err)
48 }
49 if got.ID != a.ID {
50 t.Errorf("meta.ID = %q; want %q", got.ID, a.ID)
51 }
52 if got.Task != "investigate widget" {
53 t.Errorf("meta.Task = %q", got.Task)
54 }
55 if got.Dir != a.Dir {
56 t.Errorf("meta paths drifted: %+v vs agent %+v", got, a)
57 }
58 if got.InboxPath == "" || got.EventLogPath == "" || got.SessionPath == "" {
59 t.Errorf("meta paths empty: %+v", got)
60 }
61 if got.Started.IsZero() {
62 t.Error("meta.Started is zero")
63 }
64}
65
66// TestReloadRebuildsDetachedAgents simulates a zut restart by spawning
67// in one Supervisor, throwing it away, then opening a fresh Supervisor against
68// the same root and calling Reload. The user-visible state — id,
69// task, branch, dir — must come back, and status must be Detached so
70// the dashboard can offer resume.
71func TestReloadRebuildsDetachedAgents(t *testing.T) {
72 root := t.TempDir()
73
74 // First incarnation: spawn two agents, then drop the supervisor.
75 first := New(Config{
76 Root: root,
77 RepoRoot: root,
78 NewRunner: func(a *Agent) Runner {
79 return RunnerFunc(func(ctx context.Context, _ Sink) error {
80 <-ctx.Done()
81 return ctx.Err()
82 })
83 },
84 })
85 a1, err := first.Spawn(context.Background(), "alpha task")
86 if err != nil {
87 t.Fatalf("spawn a1: %v", err)
88 }
89 a2, err := first.Spawn(context.Background(), "beta task")
90 if err != nil {
91 t.Fatalf("spawn a2: %v", err)
92 }
93 first.StopAll()
94 // Wait briefly for the runner goroutines to settle so their
95 // stop event reaches the log before we move on. Reload itself
96 // doesn't need this; it makes the assertions below deterministic.
97 a1.Wait()
98 a2.Wait()
99
100 // Second incarnation against the same root.
101 second := New(Config{
102 Root: root,
103 RepoRoot: root,
104 })
105 loaded, errs := second.Reload()
106 if len(errs) > 0 {
107 t.Fatalf("reload errs: %v", errs)
108 }
109 if loaded != 2 {
110 t.Fatalf("loaded = %d; want 2", loaded)
111 }
112 snap := second.SnapshotAll()
113 if len(snap) != 2 {
114 t.Fatalf("snapshot len = %d; want 2", len(snap))
115 }
116 ids := map[string]AgentSnapshot{}
117 for _, s := range snap {
118 ids[s.ID] = s
119 }
120 if _, ok := ids[a1.ID]; !ok {
121 t.Errorf("reloaded set missing %s", a1.ID)
122 }
123 if _, ok := ids[a2.ID]; !ok {
124 t.Errorf("reloaded set missing %s", a2.ID)
125 }
126 for _, s := range snap {
127 if s.Status != StatusDetached {
128 t.Errorf("agent %s status = %q; want detached", s.ID, s.Status)
129 }
130 if s.Task == "" {
131 t.Errorf("agent %s lost its task", s.ID)
132 }
133 }
134}
135
136func TestResumePreservesPerAgentTimeout(t *testing.T) {
137 root := t.TempDir()
138 wantTimeout := 37 * time.Minute
139 first := New(Config{
140 Root: root, RepoRoot: root,
141 NewRunner: func(a *Agent) Runner {
142 return RunnerFunc(func(ctx context.Context, _ Sink) error {
143 <-ctx.Done()
144 return ctx.Err()
145 })
146 },
147 })
148 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "timeout test", Timeout: wantTimeout})
149 if err != nil {
150 t.Fatal(err)
151 }
152 first.StopAll()
153 a.Wait()
154
155 observedCh := make(chan time.Duration, 1)
156 second := New(Config{
157 Root: root, RepoRoot: root,
158 Policy: SubagentPolicy{DefaultTimeout: time.Hour},
159 NewRunner: func(a *Agent) Runner {
160 observedCh <- a.Timeout
161 return RunnerFunc(func(ctx context.Context, _ Sink) error {
162 <-ctx.Done()
163 return ctx.Err()
164 })
165 },
166 })
167 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
168 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
169 }
170 resumed, err := second.ResumeSession(context.Background(), a.ID)
171 if err != nil {
172 t.Fatal(err)
173 }
174 var observed time.Duration
175 deadline := time.NewTimer(time.Second)
176 defer deadline.Stop()
177 select {
178 case observed = <-observedCh:
179 case <-deadline.C:
180 t.Fatal("timed out waiting for resumed runner to observe timeout")
181 }
182 if resumed.Timeout != wantTimeout || observed != wantTimeout {
183 t.Fatalf("resumed timeout = %s, runner observed %s; want %s", resumed.Timeout, observed, wantTimeout)
184 }
185 second.StopAll()
186 resumed.Wait()
187}
188
189func TestReloadRejectsMetadataPathsOutsideState(t *testing.T) {
190 root := t.TempDir()
191 outside := filepath.Join(t.TempDir(), "outside.json")
192 if err := os.WriteFile(outside, []byte("keep"), 0o600); err != nil {
193 t.Fatal(err)
194 }
195 stateDir := filepath.Join(root, "agents", "agent-1")
196 if err := os.MkdirAll(stateDir, 0o700); err != nil {
197 t.Fatal(err)
198 }
199 meta := agentMeta{
200 ID: "agent-1", Task: "malicious", Dir: root,
201 EventLogPath: outside, SessionPath: outside, InboxPath: outside,
202 }
203 data, err := json.Marshal(meta)
204 if err != nil {
205 t.Fatal(err)
206 }
207 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), data, 0o600); err != nil {
208 t.Fatal(err)
209 }
210 f := New(Config{Root: root, RepoRoot: root})
211 if loaded, errs := f.Reload(); loaded != 0 || len(errs) == 0 {
212 t.Fatalf("malicious metadata reload loaded=%d errs=%v", loaded, errs)
213 }
214 if got, err := os.ReadFile(outside); err != nil || string(got) != "keep" {
215 t.Fatalf("outside path was touched: data=%q err=%v", got, err)
216 }
217}
218
219// TestReloadIsIdempotent calls Reload twice in a row and asserts the
220// second call neither duplicates rows nor errors.
221func TestReloadIsIdempotent(t *testing.T) {
222 root := t.TempDir()
223 first := New(Config{
224 Root: root, RepoRoot: root,
225 NewRunner: func(a *Agent) Runner {
226 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
227 },
228 })
229 if _, err := first.Spawn(context.Background(), "only one"); err != nil {
230 t.Fatal(err)
231 }
232 first.StopAll()
233
234 second := New(Config{Root: root, RepoRoot: root})
235 loaded1, _ := second.Reload()
236 loaded2, errs := second.Reload()
237 if len(errs) > 0 {
238 t.Fatalf("errs on second reload: %v", errs)
239 }
240 if loaded1 != 1 || loaded2 != 0 {
241 t.Fatalf("loaded counts = (%d, %d); want (1, 0)", loaded1, loaded2)
242 }
243 if got := len(second.SnapshotAll()); got != 1 {
244 t.Fatalf("snapshot len = %d; want 1", got)
245 }
246}
247
248// TestReloadReplaysTranscriptFromEventLog drops a curated events.jsonl
249// next to a meta.json and checks that the transcript surfaces in the
250// reloaded agent. This is the user-facing payoff of Reload: opening
251// /subagents logs <id> after a restart shows what the agent said before.
252func TestReloadReplaysTranscriptFromEventLog(t *testing.T) {
253 root := t.TempDir()
254 id := "alpha-9"
255 stateDir := filepath.Join(root, "agents", id)
256 if err := os.MkdirAll(stateDir, 0o755); err != nil {
257 t.Fatal(err)
258 }
259 // meta.json
260 m := agentMeta{
261 ID: id, Task: "do thing",
262 Dir: root, Started: time.Now().Add(-time.Hour),
263 InboxPath: filepath.Join(stateDir, "in.sock"),
264 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
265 SessionPath: filepath.Join(stateDir, "session.json"),
266 }
267 mb, _ := json.MarshalIndent(m, "", " ")
268 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
269 t.Fatal(err)
270 }
271 // Event log with one assistant turn and a graceful stop.
272 log, err := OpenEventLog(m.EventLogPath)
273 if err != nil {
274 t.Fatal(err)
275 }
276 _ = log.Append(NewEvent("assistant_message", map[string]any{
277 "content": []any{map[string]any{"type": "text", "text": "hello from the past"}},
278 }))
279 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "shutdown"}))
280 _ = log.Close()
281
282 f := New(Config{Root: root, RepoRoot: root})
283 loaded, errs := f.Reload()
284 if len(errs) > 0 || loaded != 1 {
285 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
286 }
287 a := f.Get(id)
288 if a == nil {
289 t.Fatal("reloaded agent not found")
290 }
291 // Status should reflect the lifecycle terminator we wrote.
292 if a.Status() != StatusDone {
293 t.Errorf("status = %q; want done (offline)", a.Status())
294 }
295 got := a.Transcript()
296 found := false
297 for _, line := range got {
298 if line == "hello from the past" {
299 found = true
300 }
301 }
302 if !found {
303 t.Errorf("transcript missing replayed line: %v", got)
304 }
305}
306
307// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
308// doesn't blow up the whole reload. Directories with no meta.json
309// at all are silently ignored (Spawn that failed mid-way leaves
310// these); corrupt meta files are reported as errors but don't stop
311// the rest of the load.
312func TestReloadSkipsBareDirsAndCorruptMeta(t *testing.T) {
313 root := t.TempDir()
314 agentsDir := filepath.Join(root, "agents")
315 // Bare directory, no meta.json.
316 if err := os.MkdirAll(filepath.Join(agentsDir, "leftover"), 0o755); err != nil {
317 t.Fatal(err)
318 }
319 // Directory with garbage meta.json.
320 if err := os.MkdirAll(filepath.Join(agentsDir, "corrupt"), 0o755); err != nil {
321 t.Fatal(err)
322 }
323 if err := os.WriteFile(filepath.Join(agentsDir, "corrupt", "meta.json"), []byte("not json"), 0o644); err != nil {
324 t.Fatal(err)
325 }
326 // One good agent.
327 good := "good-1"
328 stateDir := filepath.Join(agentsDir, good)
329 _ = os.MkdirAll(stateDir, 0o755)
330 m := agentMeta{ID: good, Task: "x", Dir: "/tmp/x", Started: time.Now()}
331 mb, _ := json.MarshalIndent(m, "", " ")
332 _ = os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644)
333
334 f := New(Config{Root: root, RepoRoot: root})
335 loaded, errs := f.Reload()
336 if loaded != 1 {
337 t.Errorf("loaded = %d; want 1", loaded)
338 }
339 if len(errs) != 1 {
340 t.Errorf("errs len = %d; want 1 (the corrupt entry): %v", len(errs), errs)
341 }
342}
343
344// TestResumeRestartsRunnerOnSameSession is the headline test. We:
345//
346// 1. Spawn an agent with a counting runner (records each Run call).
347// 2. Stop it.
348// 3. Resume it.
349// 4. Assert Run was called twice — once for spawn, once for resume —
350// against the same SessionPath / InboxPath / Dir.
351func TestResumeRestartsRunnerOnSameSession(t *testing.T) {
352 root := t.TempDir()
353 var (
354 mu sync.Mutex
355 runs []*Agent
356 runsDone int32
357 )
358 f := New(Config{
359 Root: root, RepoRoot: root,
360 NewRunner: func(a *Agent) Runner {
361 return RunnerFunc(func(ctx context.Context, sink Sink) error {
362 mu.Lock()
363 runs = append(runs, a)
364 mu.Unlock()
365 sink.Activity("ran")
366 atomic.AddInt32(&runsDone, 1)
367 <-ctx.Done()
368 return ctx.Err()
369 })
370 },
371 })
372 defer f.StopAll()
373
374 a, err := f.Spawn(context.Background(), "do thing")
375 if err != nil {
376 t.Fatalf("spawn: %v", err)
377 }
378 originalSession := a.SessionPath
379 originalDir := a.Dir
380 originalInbox := a.InboxPath
381
382 // Wait for the runner to record its presence, then stop the agent.
383 deadline := time.Now().Add(time.Second)
384 for atomic.LoadInt32(&runsDone) < 1 && time.Now().Before(deadline) {
385 time.Sleep(2 * time.Millisecond)
386 }
387 if err := f.Stop(a.ID); err != nil {
388 t.Fatalf("stop: %v", err)
389 }
390 a.Wait()
391 if a.Status() != StatusKilled {
392 t.Fatalf("post-stop status = %q; want killed", a.Status())
393 }
394 previousAttempt := a.Snapshot().Attempt
395
396 // Now resume.
397 a2, err := f.Resume(context.Background(), a.ID)
398 if err != nil {
399 t.Fatalf("resume: %v", err)
400 }
401 if a2.ID != a.ID {
402 t.Errorf("resume returned different id: %s vs %s", a2.ID, a.ID)
403 }
404 if a2.SessionPath != originalSession {
405 t.Errorf("resume changed session path: %s vs %s", a2.SessionPath, originalSession)
406 }
407 if a2.Dir != originalDir {
408 t.Errorf("resume changed dir: %s vs %s", a2.Dir, originalDir)
409 }
410 if a2.InboxPath != originalInbox {
411 t.Errorf("resume changed inbox path: %s vs %s", a2.InboxPath, originalInbox)
412 }
413 if got := a2.Snapshot().Attempt; got != previousAttempt+1 {
414 t.Errorf("resume attempt = %d; want %d", got, previousAttempt+1)
415 }
416 // Two runner invocations: spawn + resume.
417 deadline = time.Now().Add(time.Second)
418 for atomic.LoadInt32(&runsDone) < 2 && time.Now().Before(deadline) {
419 time.Sleep(2 * time.Millisecond)
420 }
421 mu.Lock()
422 defer mu.Unlock()
423 if len(runs) != 2 {
424 t.Fatalf("runner invocations = %d; want 2", len(runs))
425 }
426 if runs[1].ID != a.ID {
427 t.Errorf("second run had wrong id: %s", runs[1].ID)
428 }
429 if got := len(f.SnapshotAll()); got != 1 {
430 t.Errorf("snapshot len = %d; want 1 (in-place replace)", got)
431 }
432}
433
434func TestResumeRecalculatesInvalidStateInboxPath(t *testing.T) {
435 if runtime.GOOS == "windows" {
436 t.Skip("subagent inbox runtime relocation applies to Unix sockets")
437 }
438 root := t.TempDir()
439 f := New(Config{
440 Root: root, RepoRoot: root,
441 NewRunner: func(*Agent) Runner {
442 return RunnerFunc(func(context.Context, Sink) error { return nil })
443 },
444 })
445 a, err := f.Spawn(context.Background(), "invalid inbox")
446 if err != nil {
447 t.Fatal(err)
448 }
449 a.Wait()
450
451 invalidPath := filepath.Join(root, "agents", a.ID, "in.sock")
452 a.InboxPath = invalidPath
453 a2, err := f.Resume(context.Background(), a.ID)
454 if err != nil {
455 t.Fatalf("resume: %v", err)
456 }
457 a2.Wait()
458 if a2.InboxPath == invalidPath {
459 t.Fatalf("resume kept state-directory inbox path %q", invalidPath)
460 }
461 if strings.HasPrefix(a2.InboxPath, root+string(filepath.Separator)) {
462 t.Fatalf("resumed inbox %q remains below state root %q", a2.InboxPath, root)
463 }
464}
465
466// TestResumeSetsResumingFlag pins the contract execRunner relies on:
467// after Resume, the new Agent has Resuming == true; after a fresh
468// Spawn it's false. The runner branches on this to decide whether
469// to pass the original Task as a positional argv to the child, so
470// getting it wrong silently re-runs the initial task on every
471// resume and surfaces "agent busy; send 'cancel' first" between
472// assistant messages.
473func TestResumeSetsResumingFlag(t *testing.T) {
474 root := t.TempDir()
475 f := New(Config{
476 Root: root, RepoRoot: root,
477 NewRunner: func(a *Agent) Runner {
478 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
479 },
480 })
481 defer f.StopAll()
482
483 a, err := f.Spawn(context.Background(), "x")
484 if err != nil {
485 t.Fatal(err)
486 }
487 if a.Resuming {
488 t.Fatal("fresh Spawn produced Resuming==true; want false")
489 }
490 if err := f.Stop(a.ID); err != nil {
491 t.Fatal(err)
492 }
493 a.Wait()
494
495 a2, err := f.Resume(context.Background(), a.ID)
496 if err != nil {
497 t.Fatalf("resume: %v", err)
498 }
499 if !a2.Resuming {
500 t.Error("Resume produced Resuming==false; want true so the runner skips the duplicate initial-task argv")
501 }
502}
503
504// TestResumeRejectsRunningAgent prevents the user from double-running
505// an agent: two runners on the same session.json would race.
506func TestResumeRejectsRunningAgent(t *testing.T) {
507 root := t.TempDir()
508 f := New(Config{
509 Root: root, RepoRoot: root,
510 NewRunner: func(a *Agent) Runner {
511 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
512 },
513 })
514 defer f.StopAll()
515 a, err := f.Spawn(context.Background(), "x")
516 if err != nil {
517 t.Fatal(err)
518 }
519 // Wait for the runner to actually start.
520 deadline := time.Now().Add(time.Second)
521 for a.Status() != StatusRunning && time.Now().Before(deadline) {
522 time.Sleep(2 * time.Millisecond)
523 }
524 if _, err := f.Resume(context.Background(), a.ID); err == nil {
525 t.Fatal("resume on running agent did not error")
526 }
527}
528
529// TestResumeAfterReload exercises the full lifecycle the user cares
530// about: spawn in process A, exit A, start process B, Reload, Resume.
531// The resumed agent must keep its id and start producing new output
532// against the same on-disk state.
533//
534// Note on transcript persistence: in production the runner is
535// execRunner, which writes every event to events.jsonl, and Reload's
536// replayEventsIntoAgent rebuilds the transcript from there. The fake
537// RunnerFunc in this test calls sink.Transcript directly (which only
538// touches in-memory state), so the "first run" line is *expected* to
539// be lost across the restart — the in-memory Agent went away. We
540// have TestReloadReplaysTranscriptFromEventLog covering the
541// log-replay path with a curated events.jsonl.
542func TestResumeAfterReload(t *testing.T) {
543 root := t.TempDir()
544 // Process A
545 a := New(Config{
546 Root: root, RepoRoot: root,
547 NewRunner: func(ag *Agent) Runner {
548 return RunnerFunc(func(ctx context.Context, sink Sink) error {
549 sink.Transcript("first run for " + ag.ID)
550 <-ctx.Done()
551 return ctx.Err()
552 })
553 },
554 })
555 ag, err := a.Spawn(context.Background(), "x")
556 if err != nil {
557 t.Fatal(err)
558 }
559 // Let the runner record its transcript line before we tear it down.
560 deadline := time.Now().Add(time.Second)
561 for len(ag.Transcript()) == 0 && time.Now().Before(deadline) {
562 time.Sleep(2 * time.Millisecond)
563 }
564 id := ag.ID
565 a.StopAll()
566 ag.Wait()
567
568 // Process B: a new Supervisor against the same root.
569 resumed := make(chan struct{}, 1)
570 b := New(Config{
571 Root: root, RepoRoot: root,
572 NewRunner: func(ag *Agent) Runner {
573 return RunnerFunc(func(ctx context.Context, sink Sink) error {
574 sink.Transcript("second run for " + ag.ID)
575 select {
576 case resumed <- struct{}{}:
577 default:
578 }
579 <-ctx.Done()
580 return ctx.Err()
581 })
582 },
583 })
584 if loaded, errs := b.Reload(); loaded != 1 || len(errs) > 0 {
585 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
586 }
587 a2, err := b.Resume(context.Background(), id)
588 if err != nil {
589 t.Fatalf("resume: %v", err)
590 }
591 defer b.StopAll()
592
593 select {
594 case <-resumed:
595 case <-time.After(time.Second):
596 t.Fatal("resume runner did not start")
597 }
598 if a2.ID != id {
599 t.Errorf("resume id = %s; want %s", a2.ID, id)
600 }
601
602 // Wait for the new transcript line to land.
603 deadline = time.Now().Add(time.Second)
604 for time.Now().Before(deadline) {
605 for _, l := range a2.Transcript() {
606 if l == "second run for "+id {
607 return
608 }
609 }
610 time.Sleep(5 * time.Millisecond)
611 }
612 t.Fatalf("resumed transcript missing fresh line: %v", a2.Transcript())
613}
614
615func TestStopMustFinishBeforeResumeOrRemove(t *testing.T) {
616 root := t.TempDir()
617 started := make(chan struct{})
618 release := make(chan struct{})
619 f := New(Config{
620 Root: root, RepoRoot: root,
621 NewRunner: func(*Agent) Runner {
622 return RunnerFunc(func(context.Context, Sink) error {
623 close(started)
624 <-release
625 return nil
626 })
627 },
628 })
629 a, err := f.Spawn(context.Background(), "slow stop")
630 if err != nil {
631 t.Fatal(err)
632 }
633 <-started
634 if err := f.Stop(a.ID); err != nil {
635 t.Fatal(err)
636 }
637 if _, err := f.Resume(context.Background(), a.ID); err == nil || !strings.Contains(err.Error(), "finalizing") {
638 t.Fatalf("immediate resume error = %v, want finalizing", err)
639 }
640 if err := f.Remove(a.ID); err == nil || !strings.Contains(err.Error(), "finalizing") {
641 t.Fatalf("immediate remove error = %v, want finalizing", err)
642 }
643 close(release)
644 a.Wait()
645}
646
647func TestResumeFencesLiveReloadedWorker(t *testing.T) {
648 if runtime.GOOS == "windows" {
649 t.Skip("subagent inbox transport uses Unix-domain sockets")
650 }
651 root := t.TempDir()
652 first := New(Config{
653 Root: root, RepoRoot: root,
654 NewRunner: func(*Agent) Runner {
655 return RunnerFunc(func(context.Context, Sink) error { return nil })
656 },
657 })
658 a, err := first.Spawn(context.Background(), "live worker")
659 if err != nil {
660 t.Fatal(err)
661 }
662 a.Wait()
663 ln, err := Listen(a.InboxPath)
664 if err != nil {
665 t.Fatal(err)
666 }
667 defer ln.Close()
668
669 second := New(Config{Root: root, RepoRoot: root})
670 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
671 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
672 }
673 if got := second.Get(a.ID).ProcessState(); got != ProcessAlive {
674 t.Fatalf("reloaded process state = %s, want alive", got)
675 }
676 if _, err := second.Resume(context.Background(), a.ID); err == nil || !strings.Contains(err.Error(), "live worker") {
677 t.Fatalf("resume error = %v, want live-worker fence", err)
678 }
679}
680
681func TestSupervisorProviderSettingsPersistAcrossReloadAndResume(t *testing.T) {
682 root := t.TempDir()
683 newSupervisor := func(baseURL string, insecureTLS bool) *Supervisor {
684 return New(Config{
685 Root: root,
686 RepoRoot: root,
687 BaseURL: baseURL,
688 InsecureTLS: insecureTLS,
689 NewRunner: func(a *Agent) Runner {
690 return RunnerFunc(func(ctx context.Context, _ Sink) error {
691 <-ctx.Done()
692 return ctx.Err()
693 })
694 },
695 })
696 }
697
698 first := newSupervisor("https://parent.example.test/v1", true)
699 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "x"})
700 if err != nil {
701 t.Fatalf("spawn: %v", err)
702 }
703 if a.BaseURL != "https://parent.example.test/v1" || !a.InsecureTLS {
704 t.Fatalf("spawned settings = (%q, %v), want inherited config", a.BaseURL, a.InsecureTLS)
705 }
706 if snap := a.Snapshot(); snap.BaseURL != a.BaseURL || snap.InsecureTLS != a.InsecureTLS {
707 t.Fatalf("snapshot settings = (%q, %v), want (%q, %v)", snap.BaseURL, snap.InsecureTLS, a.BaseURL, a.InsecureTLS)
708 }
709 if err := first.Stop(a.ID); err != nil {
710 t.Fatalf("stop: %v", err)
711 }
712 a.Wait()
713
714 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
715 if err != nil {
716 t.Fatalf("read meta.json: %v", err)
717 }
718 var meta agentMeta
719 if err := json.Unmarshal(metaBytes, &meta); err != nil {
720 t.Fatalf("decode meta.json: %v", err)
721 }
722 if meta.BaseURL != a.BaseURL || !meta.InsecureTLS {
723 t.Fatalf("metadata settings = (%q, %v), want (%q, %v)", meta.BaseURL, meta.InsecureTLS, a.BaseURL, a.InsecureTLS)
724 }
725 if strings.Contains(string(metaBytes), "api_key") || strings.Contains(string(metaBytes), "parent-secret") {
726 t.Fatalf("metadata contains credential material: %s", metaBytes)
727 }
728
729 second := newSupervisor("https://different.example.test/v1", false)
730 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
731 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
732 }
733 reloaded := second.Get(a.ID)
734 if reloaded == nil || reloaded.BaseURL != a.BaseURL || !reloaded.InsecureTLS {
735 t.Fatalf("reloaded settings = (%q, %v), want (%q, %v)", reloaded.BaseURL, reloaded.InsecureTLS, a.BaseURL, a.InsecureTLS)
736 }
737 resumed, err := second.Resume(context.Background(), a.ID)
738 if err != nil {
739 t.Fatalf("resume: %v", err)
740 }
741 if resumed.BaseURL != a.BaseURL || !resumed.InsecureTLS {
742 t.Fatalf("resumed settings = (%q, %v), want (%q, %v)", resumed.BaseURL, resumed.InsecureTLS, a.BaseURL, a.InsecureTLS)
743 }
744 if err := second.Stop(resumed.ID); err != nil {
745 t.Fatalf("stop resumed: %v", err)
746 }
747 resumed.Wait()
748}
749
750func TestExplicitChildProviderDoesNotInheritHostEndpoint(t *testing.T) {
751 f := New(Config{
752 Root: t.TempDir(), RepoRoot: t.TempDir(), Provider: "anthropic",
753 BaseURL: "https://host.example.test/v1", InsecureTLS: true,
754 NewRunner: func(*Agent) Runner {
755 return RunnerFunc(func(context.Context, Sink) error { return nil })
756 },
757 })
758 a, err := f.SpawnReq(context.Background(), SpawnRequest{Task: "x", Provider: "openai"})
759 if err != nil {
760 t.Fatal(err)
761 }
762 a.Wait()
763 if a.BaseURL != "" || a.InsecureTLS {
764 t.Fatalf("explicit provider inherited host settings: (%q, %v)", a.BaseURL, a.InsecureTLS)
765 }
766}
767
768func TestSpawnRequestProviderSettingsOverrideConfig(t *testing.T) {
769 root := t.TempDir()
770 f := New(Config{
771 Root: root,
772 RepoRoot: root,
773 BaseURL: "https://config.example.test/v1",
774 NewRunner: func(*Agent) Runner {
775 return RunnerFunc(func(context.Context, Sink) error { return nil })
776 },
777 })
778 a, err := f.SpawnReq(context.Background(), SpawnRequest{
779 Task: "x",
780 BaseURL: "https://request.example.test/v1",
781 InsecureTLS: true,
782 })
783 if err != nil {
784 t.Fatalf("spawn: %v", err)
785 }
786 a.Wait()
787 if a.BaseURL != "https://request.example.test/v1" || !a.InsecureTLS {
788 t.Fatalf("spawned settings = (%q, %v), want request override", a.BaseURL, a.InsecureTLS)
789 }
790}
791
792// TestSpawnReqPersistsModel confirms that the per-agent model
793// override is captured at Spawn time, surfaced via Snapshot, and
794// written to meta.json so a later Reload + Resume reuses it.
795func TestSpawnReqPersistsModel(t *testing.T) {
796 root := t.TempDir()
797 f := New(Config{
798 Root: root, RepoRoot: root,
799 NewRunner: func(a *Agent) Runner {
800 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
801 },
802 })
803 f.SetActiveSession("host-session")
804 a, err := f.SpawnReq(context.Background(), SpawnRequest{
805 Task: "x", Model: "claude-sonnet-4-5", Provider: "anthropic",
806 Reasoning: "high", Subagent: "reviewer",
807 })
808 if err != nil {
809 t.Fatalf("spawn: %v", err)
810 }
811 if a.Model != "claude-sonnet-4-5" || a.Provider != "anthropic" || a.Reasoning != "high" || a.Subagent != "reviewer" || a.SessionID != "host-session" {
812 t.Fatalf("agent fields = (%q,%q,%q,%q,%q); want model/provider/reasoning/profile/session", a.Model, a.Provider, a.Reasoning, a.Subagent, a.SessionID)
813 }
814 snap := a.Snapshot()
815 if snap.Model != "claude-sonnet-4-5" || snap.Provider != "anthropic" || snap.Reasoning != "high" || snap.Subagent != "reviewer" {
816 t.Fatalf("snapshot = (%q,%q,%q,%q); child fields not surfaced", snap.Model, snap.Provider, snap.Reasoning, snap.Subagent)
817 }
818
819 // Stop so we can read meta.json without racing the run loop.
820 if err := f.Stop(a.ID); err != nil {
821 t.Fatal(err)
822 }
823 a.Wait()
824
825 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
826 if err != nil {
827 t.Fatalf("read meta.json: %v", err)
828 }
829 var got agentMeta
830 if err := json.Unmarshal(metaBytes, &got); err != nil {
831 t.Fatal(err)
832 }
833 if got.Model != "claude-sonnet-4-5" || got.Provider != "anthropic" || got.Reasoning != "high" || got.Subagent != "reviewer" || got.SessionID != "host-session" {
834 t.Errorf("meta = (%q,%q,%q,%q,%q); want model/provider/reasoning/profile/session persisted", got.Model, got.Provider, got.Reasoning, got.Subagent, got.SessionID)
835 }
836
837 // Reload in a fresh Supervisor and confirm the detached agent still
838 // carries the model/provider so Resume can route the child
839 // subprocess back to the same model.
840 g := New(Config{
841 Root: root, RepoRoot: root,
842 NewRunner: func(a *Agent) Runner {
843 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
844 },
845 })
846 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
847 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
848 }
849 re := g.Get(a.ID)
850 if re == nil {
851 t.Fatal("reloaded agent missing")
852 }
853 if re.Model != "claude-sonnet-4-5" || re.Provider != "anthropic" || re.Reasoning != "high" || re.Subagent != "reviewer" || re.SessionID != "host-session" {
854 t.Errorf("reloaded fields = (%q,%q,%q,%q,%q); want preserved", re.Model, re.Provider, re.Reasoning, re.Subagent, re.SessionID)
855 }
856
857 resumed, err := g.Resume(context.Background(), re.ID)
858 if err != nil {
859 t.Fatalf("resume: %v", err)
860 }
861 if resumed.Reasoning != "high" || resumed.Subagent != "reviewer" || resumed.SessionID != "host-session" {
862 t.Errorf("resumed fields = (%q,%q,%q); want reasoning/profile/session preserved", resumed.Reasoning, resumed.Subagent, resumed.SessionID)
863 }
864 if err := g.Stop(resumed.ID); err != nil {
865 t.Fatalf("stop resumed agent: %v", err)
866 }
867 resumed.Wait()
868}
869
870// TestSubagentWorkerArgsIncludesModelFlags pins the argv contract that
871// connects the supervisor's per-agent model override to the child's
872// --model / --provider flag set. Adding a new path to argv without
873// updating this assertion is the failure mode this catches.
874func TestSubagentWorkerArgsIncludesModelFlags(t *testing.T) {
875 args := subagentWorkerArgs(subagentWorkerArgsOpts{
876 Exe: "/zut", Dir: "/wt", SessionPath: "/s.json", InboxPath: "/in.sock",
877 Task: "do x", Model: "gpt-5", Provider: "openai",
878 })
879 want := []string{"--model", "gpt-5", "--provider", "openai"}
880 for i := 0; i+1 < len(want); i += 2 {
881 flag, value := want[i], want[i+1]
882 found := false
883 for j := 0; j+1 < len(args); j++ {
884 if args[j] == flag && args[j+1] == value {
885 found = true
886 break
887 }
888 }
889 if !found {
890 t.Errorf("argv missing %s %s: %v", flag, value, args)
891 }
892 }
893 // And the task still ends up positional last.
894 if args[len(args)-1] != "do x" {
895 t.Errorf("task should be last positional; argv=%v", args)
896 }
897}
898
899// TestStopOnDetachedAgentIsNoopAndDoesNotPanic regression-tests the
900// segfault that hit when the user pressed 'k' on a reloaded
901// (detached) agent: Supervisor.Stop unconditionally called a.cancel(),
902// but buildDetachedAgent never assigns a cancel func because there's
903// no in-process runner to cancel. The fix is to short-circuit Stop
904// for StatusDetached and — belt-and-braces — nil-check a.cancel.
905func TestStopOnDetachedAgentIsNoopAndDoesNotPanic(t *testing.T) {
906 root := t.TempDir()
907 // Build a detached agent the same way Reload does: drop a
908 // meta.json on disk, then ask a fresh Supervisor to pick it up.
909 id := "alpha-1"
910 stateDir := filepath.Join(root, "agents", id)
911 if err := os.MkdirAll(stateDir, 0o755); err != nil {
912 t.Fatal(err)
913 }
914 m := agentMeta{
915 ID: id, Task: "t",
916 Dir: root,
917 Started: time.Now().Add(-time.Hour),
918 InboxPath: filepath.Join(stateDir, "in.sock"),
919 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
920 SessionPath: filepath.Join(stateDir, "session.json"),
921 }
922 mb, _ := json.MarshalIndent(m, "", " ")
923 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
924 t.Fatal(err)
925 }
926
927 f := New(Config{Root: root, RepoRoot: root})
928 if loaded, errs := f.Reload(); loaded != 1 || len(errs) > 0 {
929 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
930 }
931 a := f.Get(id)
932 if a == nil {
933 t.Fatal("detached agent missing from subagent")
934 }
935 if a.Status() != StatusDetached {
936 t.Fatalf("setup: agent status = %q; want detached", a.Status())
937 }
938
939 // The real assertion: Stop returns cleanly without panicking.
940 defer func() {
941 if r := recover(); r != nil {
942 t.Fatalf("Stop on detached agent panicked: %v", r)
943 }
944 }()
945 if err := f.Stop(id); err != nil {
946 t.Fatalf("Stop on detached agent: %v", err)
947 }
948 // Status must remain Detached (Stop is a no-op).
949 if got := a.Status(); got != StatusDetached {
950 t.Errorf("after Stop status = %q; want detached (no-op)", got)
951 }
952
953 // StopAll, which the cli defers on exit, must also be safe.
954 f.StopAll()
955}
956
957// TestRemoveAlsoCleansStateDir ensures Remove deletes the on-disk
958// state directory in addition to the worktree, so a removed agent
959// doesn't reappear on the next Reload.
960func TestRemoveAlsoCleansStateDir(t *testing.T) {
961 root := t.TempDir()
962 f := New(Config{
963 Root: root, RepoRoot: root,
964 NewRunner: func(a *Agent) Runner {
965 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
966 },
967 })
968 a, err := f.Spawn(context.Background(), "x")
969 if err != nil {
970 t.Fatal(err)
971 }
972 stateDir := filepath.Join(root, "agents", a.ID)
973 if _, err := os.Stat(filepath.Join(stateDir, "meta.json")); err != nil {
974 t.Fatalf("meta.json missing pre-remove: %v", err)
975 }
976 if err := f.Stop(a.ID); err != nil {
977 t.Fatal(err)
978 }
979 a.Wait()
980 if err := f.Remove(a.ID); err != nil {
981 t.Fatal(err)
982 }
983 if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
984 t.Fatalf("state dir should be gone after remove; got %v", err)
985 }
986
987 // A fresh Supervisor + Reload should find nothing.
988 g := New(Config{Root: root, RepoRoot: root})
989 if loaded, _ := g.Reload(); loaded != 0 {
990 t.Fatalf("reload after remove loaded=%d; want 0", loaded)
991 }
992}
993
994func TestRemoveReloadedAgentKeepsSiblingState(t *testing.T) {
995 root := t.TempDir()
996 first := New(Config{
997 Root: root, RepoRoot: root,
998 NewRunner: func(*Agent) Runner {
999 return RunnerFunc(func(context.Context, Sink) error { return nil })
1000 },
1001 })
1002 a1, err := first.Spawn(context.Background(), "first")
1003 if err != nil {
1004 t.Fatal(err)
1005 }
1006 a2, err := first.Spawn(context.Background(), "second")
1007 if err != nil {
1008 t.Fatal(err)
1009 }
1010 a1.Wait()
1011 a2.Wait()
1012
1013 second := New(Config{Root: root, RepoRoot: root})
1014 if loaded, errs := second.Reload(); loaded != 2 || len(errs) != 0 {
1015 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1016 }
1017 if err := second.Remove(a1.ID); err != nil {
1018 t.Fatal(err)
1019 }
1020 if _, err := os.Stat(filepath.Join(root, "agents", a2.ID, "meta.json")); err != nil {
1021 t.Fatalf("removing reloaded sibling deleted surviving metadata: %v", err)
1022 }
1023}
1024
1025// TestActiveSessionScopesSnapshotAll proves the session filter the
1026// user asked for: a single Supervisor with agents spawned under two
1027// different active sessions only surfaces the agents matching the
1028// currently-active session via SnapshotAll. Switching the active
1029// session re-narrows the view without touching agent state.
1030func TestActiveSessionScopesSnapshotAll(t *testing.T) {
1031 root := t.TempDir()
1032 f := New(Config{
1033 Root: root, RepoRoot: root,
1034 NewRunner: func(a *Agent) Runner {
1035 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1036 },
1037 })
1038
1039 // Spawn one agent under session A and one under session B.
1040 f.SetActiveSession("sess-A")
1041 aA, err := f.Spawn(context.Background(), "task A")
1042 if err != nil {
1043 t.Fatal(err)
1044 }
1045 f.SetActiveSession("sess-B")
1046 aB, err := f.Spawn(context.Background(), "task B")
1047 if err != nil {
1048 t.Fatal(err)
1049 }
1050
1051 // Both agents must carry the session id they were spawned under,
1052 // regardless of any later SetActiveSession.
1053 if aA.SessionID != "sess-A" {
1054 t.Errorf("aA.SessionID = %q; want sess-A", aA.SessionID)
1055 }
1056 if aB.SessionID != "sess-B" {
1057 t.Errorf("aB.SessionID = %q; want sess-B", aB.SessionID)
1058 }
1059
1060 // Active session B → only aB visible.
1061 only := snapshotIDs(f.SnapshotAll())
1062 if len(only) != 1 || only[0] != aB.ID {
1063 t.Errorf("scoped to sess-B, snapshot ids = %v; want [%s]", only, aB.ID)
1064 }
1065
1066 // Switch back to A → only aA visible.
1067 f.SetActiveSession("sess-A")
1068 only = snapshotIDs(f.SnapshotAll())
1069 if len(only) != 1 || only[0] != aA.ID {
1070 t.Errorf("scoped to sess-A, snapshot ids = %v; want [%s]", only, aA.ID)
1071 }
1072
1073 // Clear the scope → both visible.
1074 f.SetActiveSession("")
1075 all := snapshotIDs(f.SnapshotAll())
1076 if len(all) != 2 {
1077 t.Errorf("unscoped snapshot ids = %v; want both agents", all)
1078 }
1079
1080 // Cleanup.
1081 _ = f.Stop(aA.ID)
1082 _ = f.Stop(aB.ID)
1083 aA.Wait()
1084 aB.Wait()
1085}
1086
1087// TestSessionIDPersistsAcrossReload confirms the SessionID field
1088// survives a Supervisor restart: an agent spawned under session A in one
1089// Supervisor instance is still SessionID="sess-A" after a fresh New +
1090// Reload reads it back from meta.json. Without persistence, the
1091// scope filter would forget which session owned each agent after
1092// a zut restart and the dashboard would leak everything again.
1093func TestSessionIDPersistsAcrossReload(t *testing.T) {
1094 root := t.TempDir()
1095 mkSupervisor := func() *Supervisor {
1096 return New(Config{
1097 Root: root, RepoRoot: root,
1098 NewRunner: func(a *Agent) Runner {
1099 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1100 },
1101 })
1102 }
1103
1104 f := mkSupervisor()
1105 f.SetActiveSession("sess-keep")
1106 a, err := f.Spawn(context.Background(), "persist me")
1107 if err != nil {
1108 t.Fatal(err)
1109 }
1110 _ = f.Stop(a.ID)
1111 a.Wait()
1112
1113 // Fresh Supervisor, same root, reload from disk.
1114 g := mkSupervisor()
1115 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
1116 t.Fatalf("reload loaded=%d errs=%v; want 1 / no errs", loaded, errs)
1117 }
1118 g.SetActiveSession("sess-keep")
1119 got := snapshotIDs(g.SnapshotAll())
1120 if len(got) != 1 || got[0] != a.ID {
1121 t.Errorf("after reload + scope to sess-keep, ids = %v; want [%s]", got, a.ID)
1122 }
1123
1124 // A terminal persisted agent remains visible from another host session
1125 // so a normal restart does not hide durable recovery state.
1126 g.SetActiveSession("sess-other")
1127 if got := snapshotIDs(g.SnapshotAll()); len(got) != 1 || got[0] != a.ID {
1128 t.Errorf("scoped to other session, ids = %v; want [%s]", got, a.ID)
1129 }
1130}
1131
1132// TestEmptySessionIDIsVisibleFromAnyScope pins the unscoped-agent
1133// rule: agents whose meta.json predates the SessionID field (or who
1134// were spawned without an active session, e.g. via a test rig or
1135// scripted caller) carry SessionID == "" and remain visible from
1136// every scope. Otherwise the schema bump would orphan every
1137// pre-existing agent the moment a user upgraded zut.
1138func TestEmptySessionIDIsVisibleFromAnyScope(t *testing.T) {
1139 root := t.TempDir()
1140 f := New(Config{
1141 Root: root, RepoRoot: root,
1142 NewRunner: func(a *Agent) Runner {
1143 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1144 },
1145 })
1146 // No SetActiveSession call → agent spawned with empty SessionID.
1147 a, err := f.Spawn(context.Background(), "unscoped")
1148 if err != nil {
1149 t.Fatal(err)
1150 }
1151 if a.SessionID != "" {
1152 t.Fatalf("unscoped spawn produced SessionID %q; want empty", a.SessionID)
1153 }
1154
1155 for _, scope := range []string{"", "any-session", "some-other"} {
1156 f.SetActiveSession(scope)
1157 got := snapshotIDs(f.SnapshotAll())
1158 if len(got) != 1 || got[0] != a.ID {
1159 t.Errorf("scope=%q: ids = %v; want unscoped agent visible", scope, got)
1160 }
1161 }
1162
1163 _ = f.Stop(a.ID)
1164 a.Wait()
1165}
1166
1167func snapshotIDs(ss []AgentSnapshot) []string {
1168 out := make([]string, len(ss))
1169 for i, s := range ss {
1170 out[i] = s.ID
1171 }
1172 return out
1173}