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
17func TestWriteAgentMetaSerializesConcurrentSnapshots(t *testing.T) {
18 stateDir := t.TempDir()
19 a := &Agent{ID: "concurrent-meta", Task: "retain queued prompts", stateDir: stateDir}
20 var wg sync.WaitGroup
21 for i := 0; i < 32; i++ {
22 wg.Add(1)
23 go func(i int) {
24 defer wg.Done()
25 a.setTurnCounts(i, i/2)
26 a.queueResumePrompt("prompt", time.Now())
27 if err := writeAgentMeta(stateDir, a); err != nil {
28 t.Errorf("write metadata: %v", err)
29 }
30 }(i)
31 }
32 wg.Wait()
33 if _, err := readAgentMeta(stateDir); err != nil {
34 t.Fatalf("final metadata is not a complete JSON snapshot: %v", err)
35 }
36 matches, err := filepath.Glob(filepath.Join(stateDir, ".meta.json.tmp-*"))
37 if err != nil {
38 t.Fatal(err)
39 }
40 if len(matches) != 0 {
41 t.Fatalf("temporary metadata files remain: %v", matches)
42 }
43}
44
45// TestSpawnWritesMetaJSON asserts the durability contract: every
46// successful Spawn leaves a meta.json on disk with the agent's
47// identity bits. Without this, Reload on the next zut launch can't
48// find the agent and the user loses access to the worktree.
49func TestSpawnWritesMetaJSON(t *testing.T) {
50 root := t.TempDir()
51 f := New(Config{
52 Root: root,
53 RepoRoot: root,
54 WebSearchPolicy: WebSearchAllow,
55 NewRunner: func(a *Agent) Runner {
56 return RunnerFunc(func(ctx context.Context, _ Sink) error {
57 <-ctx.Done()
58 return ctx.Err()
59 })
60 },
61 })
62 defer f.StopAll()
63
64 a, err := f.Spawn(context.Background(), "investigate widget")
65 if err != nil {
66 t.Fatalf("spawn: %v", err)
67 }
68
69 // Read meta.json straight off disk.
70 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
71 if err != nil {
72 t.Fatalf("read meta.json: %v", err)
73 }
74 var got agentMeta
75 if err := json.Unmarshal(metaBytes, &got); err != nil {
76 t.Fatalf("parse meta.json: %v", err)
77 }
78 if got.ID != a.ID {
79 t.Errorf("meta.ID = %q; want %q", got.ID, a.ID)
80 }
81 if got.Task != "investigate widget" {
82 t.Errorf("meta.Task = %q", got.Task)
83 }
84 if got.Dir != a.Dir {
85 t.Errorf("meta paths drifted: %+v vs agent %+v", got, a)
86 }
87 if got.InboxPath == "" || got.EventLogPath == "" || got.SessionPath == "" {
88 t.Errorf("meta paths empty: %+v", got)
89 }
90 if got.Started.IsZero() {
91 t.Error("meta.Started is zero")
92 }
93 if got.WebSearchPolicy != WebSearchAllow || a.WebSearchPolicy != WebSearchAllow {
94 t.Fatalf("web search policy persisted as meta=%v agent=%v, want allow", got.WebSearchPolicy, a.WebSearchPolicy)
95 }
96}
97
98// TestReloadRebuildsDetachedAgents simulates a zut restart by spawning
99// in one Supervisor, throwing it away, then opening a fresh Supervisor against
100// the same root and calling Reload. The user-visible state — id,
101// task, branch, dir — must come back, and status must be Detached so
102// the dashboard can offer resume.
103func TestReloadRebuildsDetachedAgents(t *testing.T) {
104 root := t.TempDir()
105
106 // First incarnation: spawn two agents, then drop the supervisor.
107 first := New(Config{
108 Root: root,
109 RepoRoot: root,
110 NewRunner: func(a *Agent) Runner {
111 return RunnerFunc(func(ctx context.Context, _ Sink) error {
112 <-ctx.Done()
113 return ctx.Err()
114 })
115 },
116 })
117 a1, err := first.Spawn(context.Background(), "alpha task")
118 if err != nil {
119 t.Fatalf("spawn a1: %v", err)
120 }
121 a2, err := first.Spawn(context.Background(), "beta task")
122 if err != nil {
123 t.Fatalf("spawn a2: %v", err)
124 }
125 first.StopAll()
126 // Wait briefly for the runner goroutines to settle so their
127 // stop event reaches the log before we move on. Reload itself
128 // doesn't need this; it makes the assertions below deterministic.
129 a1.Wait()
130 a2.Wait()
131
132 // Second incarnation against the same root.
133 second := New(Config{
134 Root: root,
135 RepoRoot: root,
136 })
137 loaded, errs := second.Reload()
138 if len(errs) > 0 {
139 t.Fatalf("reload errs: %v", errs)
140 }
141 if loaded != 2 {
142 t.Fatalf("loaded = %d; want 2", loaded)
143 }
144 snap := second.SnapshotAll()
145 if len(snap) != 2 {
146 t.Fatalf("snapshot len = %d; want 2", len(snap))
147 }
148 ids := map[string]AgentSnapshot{}
149 for _, s := range snap {
150 ids[s.ID] = s
151 }
152 if _, ok := ids[a1.ID]; !ok {
153 t.Errorf("reloaded set missing %s", a1.ID)
154 }
155 if _, ok := ids[a2.ID]; !ok {
156 t.Errorf("reloaded set missing %s", a2.ID)
157 }
158 for _, s := range snap {
159 if s.Status != StatusDetached {
160 t.Errorf("agent %s status = %q; want detached", s.ID, s.Status)
161 }
162 if s.Task == "" {
163 t.Errorf("agent %s lost its task", s.ID)
164 }
165 }
166}
167
168func TestResumePreservesPerAgentTimeout(t *testing.T) {
169 root := t.TempDir()
170 wantTimeout := 37 * time.Minute
171 first := New(Config{
172 Root: root, RepoRoot: root,
173 NewRunner: func(a *Agent) Runner {
174 return RunnerFunc(func(ctx context.Context, _ Sink) error {
175 <-ctx.Done()
176 return ctx.Err()
177 })
178 },
179 })
180 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "timeout test", Timeout: wantTimeout})
181 if err != nil {
182 t.Fatal(err)
183 }
184 first.StopAll()
185 a.Wait()
186
187 observedCh := make(chan time.Duration, 1)
188 second := New(Config{
189 Root: root, RepoRoot: root,
190 Policy: SubagentPolicy{DefaultTimeout: time.Hour},
191 NewRunner: func(a *Agent) Runner {
192 observedCh <- a.Timeout
193 return RunnerFunc(func(ctx context.Context, _ Sink) error {
194 <-ctx.Done()
195 return ctx.Err()
196 })
197 },
198 })
199 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
200 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
201 }
202 resumed, err := second.ResumeSession(context.Background(), a.ID)
203 if err != nil {
204 t.Fatal(err)
205 }
206 var observed time.Duration
207 deadline := time.NewTimer(time.Second)
208 defer deadline.Stop()
209 select {
210 case observed = <-observedCh:
211 case <-deadline.C:
212 t.Fatal("timed out waiting for resumed runner to observe timeout")
213 }
214 if resumed.Timeout != wantTimeout || observed != wantTimeout {
215 t.Fatalf("resumed timeout = %s, runner observed %s; want %s", resumed.Timeout, observed, wantTimeout)
216 }
217 second.StopAll()
218 resumed.Wait()
219}
220
221func TestReloadRejectsMetadataPathsOutsideState(t *testing.T) {
222 root := t.TempDir()
223 outside := filepath.Join(t.TempDir(), "outside.json")
224 if err := os.WriteFile(outside, []byte("keep"), 0o600); err != nil {
225 t.Fatal(err)
226 }
227 stateDir := filepath.Join(root, "agents", "agent-1")
228 if err := os.MkdirAll(stateDir, 0o700); err != nil {
229 t.Fatal(err)
230 }
231 meta := agentMeta{
232 ID: "agent-1", Task: "malicious", Dir: root,
233 EventLogPath: outside, SessionPath: outside, InboxPath: outside,
234 }
235 data, err := json.Marshal(meta)
236 if err != nil {
237 t.Fatal(err)
238 }
239 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), data, 0o600); err != nil {
240 t.Fatal(err)
241 }
242 f := New(Config{Root: root, RepoRoot: root})
243 if loaded, errs := f.Reload(); loaded != 0 || len(errs) == 0 {
244 t.Fatalf("malicious metadata reload loaded=%d errs=%v", loaded, errs)
245 }
246 if got, err := os.ReadFile(outside); err != nil || string(got) != "keep" {
247 t.Fatalf("outside path was touched: data=%q err=%v", got, err)
248 }
249}
250
251// TestReloadIsIdempotent calls Reload twice in a row and asserts the
252// second call neither duplicates rows nor errors.
253func TestReloadIsIdempotent(t *testing.T) {
254 root := t.TempDir()
255 first := New(Config{
256 Root: root, RepoRoot: root,
257 NewRunner: func(a *Agent) Runner {
258 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
259 },
260 })
261 if _, err := first.Spawn(context.Background(), "only one"); err != nil {
262 t.Fatal(err)
263 }
264 first.StopAll()
265
266 second := New(Config{Root: root, RepoRoot: root})
267 loaded1, _ := second.Reload()
268 loaded2, errs := second.Reload()
269 if len(errs) > 0 {
270 t.Fatalf("errs on second reload: %v", errs)
271 }
272 if loaded1 != 1 || loaded2 != 0 {
273 t.Fatalf("loaded counts = (%d, %d); want (1, 0)", loaded1, loaded2)
274 }
275 if got := len(second.SnapshotAll()); got != 1 {
276 t.Fatalf("snapshot len = %d; want 1", got)
277 }
278}
279
280// TestReloadReplaysTranscriptFromEventLog drops a curated events.jsonl
281// next to a meta.json and checks that the transcript surfaces in the
282// reloaded agent. This is the user-facing payoff of Reload: opening
283// /subagents logs <id> after a restart shows what the agent said before.
284func TestReloadReplaysTranscriptFromEventLog(t *testing.T) {
285 root := t.TempDir()
286 id := "alpha-9"
287 stateDir := filepath.Join(root, "agents", id)
288 if err := os.MkdirAll(stateDir, 0o755); err != nil {
289 t.Fatal(err)
290 }
291 // meta.json
292 m := agentMeta{
293 ID: id, Task: "do thing",
294 Dir: root, Started: time.Now().Add(-time.Hour),
295 InboxPath: filepath.Join(stateDir, "in.sock"),
296 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
297 SessionPath: filepath.Join(stateDir, "session.json"),
298 }
299 mb, _ := json.MarshalIndent(m, "", " ")
300 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
301 t.Fatal(err)
302 }
303 // Event log with one assistant turn and a graceful stop.
304 log, err := OpenEventLog(m.EventLogPath)
305 if err != nil {
306 t.Fatal(err)
307 }
308 _ = log.Append(NewEvent("assistant_message", map[string]any{
309 "content": []any{map[string]any{"type": "text", "text": "hello from the past"}},
310 }))
311 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "shutdown"}))
312 _ = log.Close()
313
314 f := New(Config{Root: root, RepoRoot: root})
315 loaded, errs := f.Reload()
316 if len(errs) > 0 || loaded != 1 {
317 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
318 }
319 a := f.Get(id)
320 if a == nil {
321 t.Fatal("reloaded agent not found")
322 }
323 // Legacy metadata has no persisted dashboard status, so startup keeps the
324 // agent detached until the user asks to inspect its durable history.
325 if a.Status() != StatusDetached {
326 t.Errorf("startup status = %q; want detached", a.Status())
327 }
328 got := a.Transcript()
329 if a.Status() != StatusDone {
330 t.Errorf("hydrated status = %q; want done (offline)", a.Status())
331 }
332 found := false
333 for _, line := range got {
334 if line == "hello from the past" {
335 found = true
336 }
337 }
338 if !found {
339 t.Errorf("transcript missing replayed line: %v", got)
340 }
341}
342
343// TestReloadUsesPersistedSummaryBeforeTranscriptHydration ensures startup
344// rebuilds the dashboard from compact metadata rather than replaying the
345// event log. Opening the transcript remains the explicit point that loads
346// durable history.
347func TestReloadUsesPersistedSummaryBeforeTranscriptHydration(t *testing.T) {
348 root := t.TempDir()
349 id := "summary-1"
350 stateDir := filepath.Join(root, "agents", id)
351 agent := &Agent{
352 ID: id,
353 Task: "persist summary",
354 OriginalTask: "persist summary",
355 Dir: root,
356 Started: time.Now().Add(-time.Hour),
357 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
358 SessionPath: filepath.Join(stateDir, "session.json"),
359 status: StatusDone,
360 activity: "done",
361 finished: time.Now().Add(-time.Minute),
362 processState: ProcessExited,
363 turnState: TurnSucceeded,
364 currentTurnID: "turn-1",
365 done: make(chan struct{}),
366 }
367 if err := writeAgentMeta(stateDir, agent); err != nil {
368 t.Fatal(err)
369 }
370
371 log, err := OpenEventLog(agent.EventLogPath)
372 if err != nil {
373 t.Fatal(err)
374 }
375 if err := log.Append(NewEvent("assistant_message", map[string]any{
376 "content": []any{map[string]any{"type": "text", "text": "historical reply"}},
377 })); err != nil {
378 t.Fatal(err)
379 }
380 if err := log.Append(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"})); err != nil {
381 t.Fatal(err)
382 }
383 if err := log.Close(); err != nil {
384 t.Fatal(err)
385 }
386
387 supervisor := New(Config{Root: root, RepoRoot: root})
388 if loaded, errs := supervisor.Reload(); loaded != 1 || len(errs) != 0 {
389 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
390 }
391 reloaded := supervisor.Get(id)
392 if reloaded == nil {
393 t.Fatal("reloaded agent missing")
394 }
395 before := reloaded.Snapshot()
396 if before.Status != StatusDetached || before.Activity != "done" || before.ProcessState != ProcessExited || before.TurnState != TurnSucceeded {
397 t.Fatalf("startup summary = (%q, %q, %q, %q), want (detached, done, exited, succeeded)", before.Status, before.Activity, before.ProcessState, before.TurnState)
398 }
399 if len(before.Lines) != 0 {
400 t.Fatalf("startup loaded transcript lines: %v", before.Lines)
401 }
402 lastActivity := before.LastActivity
403
404 lines := reloaded.Transcript()
405 if len(lines) != 1 || lines[0] != "historical reply" {
406 t.Fatalf("hydrated transcript = %v", lines)
407 }
408 after := reloaded.Snapshot()
409 if after.Status != StatusDetached || after.Activity != "done" {
410 t.Fatalf("hydration overwrote persisted summary: (%q, %q)", after.Status, after.Activity)
411 }
412 if !after.LastActivity.Equal(lastActivity) {
413 t.Fatalf("hydration changed last activity from %s to %s", lastActivity, after.LastActivity)
414 }
415}
416
417// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
418// doesn't blow up the whole reload. Directories with no meta.json
419// at all are silently ignored (Spawn that failed mid-way leaves
420// these); corrupt meta files are reported as errors but don't stop
421// the rest of the load.
422func TestReloadSkipsBareDirsAndCorruptMeta(t *testing.T) {
423 root := t.TempDir()
424 agentsDir := filepath.Join(root, "agents")
425 // Bare directory, no meta.json.
426 if err := os.MkdirAll(filepath.Join(agentsDir, "leftover"), 0o755); err != nil {
427 t.Fatal(err)
428 }
429 // Directory with garbage meta.json.
430 if err := os.MkdirAll(filepath.Join(agentsDir, "corrupt"), 0o755); err != nil {
431 t.Fatal(err)
432 }
433 if err := os.WriteFile(filepath.Join(agentsDir, "corrupt", "meta.json"), []byte("not json"), 0o644); err != nil {
434 t.Fatal(err)
435 }
436 // One good agent.
437 good := "good-1"
438 stateDir := filepath.Join(agentsDir, good)
439 _ = os.MkdirAll(stateDir, 0o755)
440 m := agentMeta{ID: good, Task: "x", Dir: "/tmp/x", Started: time.Now()}
441 mb, _ := json.MarshalIndent(m, "", " ")
442 _ = os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644)
443
444 f := New(Config{Root: root, RepoRoot: root})
445 loaded, errs := f.Reload()
446 if loaded != 1 {
447 t.Errorf("loaded = %d; want 1", loaded)
448 }
449 if len(errs) != 1 {
450 t.Errorf("errs len = %d; want 1 (the corrupt entry): %v", len(errs), errs)
451 }
452}
453
454// TestResumeRestartsRunnerOnSameSession is the headline test. We:
455//
456// 1. Spawn an agent with a counting runner (records each Run call).
457// 2. Stop it.
458// 3. Resume it.
459// 4. Assert Run was called twice — once for spawn, once for resume —
460// against the same SessionPath / InboxPath / Dir.
461func TestResumeRestartsRunnerOnSameSession(t *testing.T) {
462 root := t.TempDir()
463 var (
464 mu sync.Mutex
465 runs []*Agent
466 runsDone int32
467 )
468 f := New(Config{
469 Root: root, RepoRoot: root,
470 NewRunner: func(a *Agent) Runner {
471 return RunnerFunc(func(ctx context.Context, sink Sink) error {
472 mu.Lock()
473 runs = append(runs, a)
474 mu.Unlock()
475 sink.Activity("ran")
476 atomic.AddInt32(&runsDone, 1)
477 <-ctx.Done()
478 return ctx.Err()
479 })
480 },
481 })
482 defer f.StopAll()
483
484 a, err := f.Spawn(context.Background(), "do thing")
485 if err != nil {
486 t.Fatalf("spawn: %v", err)
487 }
488 originalSession := a.SessionPath
489 originalDir := a.Dir
490 originalInbox := a.InboxPath
491
492 // Wait for the runner to record its presence, then stop the agent.
493 deadline := time.Now().Add(time.Second)
494 for atomic.LoadInt32(&runsDone) < 1 && time.Now().Before(deadline) {
495 time.Sleep(2 * time.Millisecond)
496 }
497 if err := f.Stop(a.ID); err != nil {
498 t.Fatalf("stop: %v", err)
499 }
500 a.Wait()
501 if a.Status() != StatusKilled {
502 t.Fatalf("post-stop status = %q; want killed", a.Status())
503 }
504 previousAttempt := a.Snapshot().Attempt
505
506 // Now resume.
507 a2, err := f.Resume(context.Background(), a.ID)
508 if err != nil {
509 t.Fatalf("resume: %v", err)
510 }
511 if a2.ID != a.ID {
512 t.Errorf("resume returned different id: %s vs %s", a2.ID, a.ID)
513 }
514 if a2.SessionPath != originalSession {
515 t.Errorf("resume changed session path: %s vs %s", a2.SessionPath, originalSession)
516 }
517 if a2.Dir != originalDir {
518 t.Errorf("resume changed dir: %s vs %s", a2.Dir, originalDir)
519 }
520 if a2.InboxPath != originalInbox {
521 t.Errorf("resume changed inbox path: %s vs %s", a2.InboxPath, originalInbox)
522 }
523 if got := a2.Snapshot().Attempt; got != previousAttempt+1 {
524 t.Errorf("resume attempt = %d; want %d", got, previousAttempt+1)
525 }
526 // Two runner invocations: spawn + resume.
527 deadline = time.Now().Add(time.Second)
528 for atomic.LoadInt32(&runsDone) < 2 && time.Now().Before(deadline) {
529 time.Sleep(2 * time.Millisecond)
530 }
531 mu.Lock()
532 defer mu.Unlock()
533 if len(runs) != 2 {
534 t.Fatalf("runner invocations = %d; want 2", len(runs))
535 }
536 if runs[1].ID != a.ID {
537 t.Errorf("second run had wrong id: %s", runs[1].ID)
538 }
539 if got := len(f.SnapshotAll()); got != 1 {
540 t.Errorf("snapshot len = %d; want 1 (in-place replace)", got)
541 }
542}
543
544func TestRestartTaskRetainsQueuedResumePrompt(t *testing.T) {
545 root := t.TempDir()
546 started := make(chan *Agent, 2)
547 f := New(Config{
548 Root: root, RepoRoot: root,
549 NewRunner: func(a *Agent) Runner {
550 return RunnerFunc(func(ctx context.Context, _ Sink) error {
551 started <- a
552 <-ctx.Done()
553 return ctx.Err()
554 })
555 },
556 })
557 t.Cleanup(f.StopAll)
558
559 existing, err := f.Spawn(context.Background(), "review the implementation")
560 if err != nil {
561 t.Fatal(err)
562 }
563 select {
564 case got := <-started:
565 if got != existing {
566 t.Fatalf("initial runner agent = %p, want %p", got, existing)
567 }
568 case <-time.After(time.Second):
569 t.Fatal("initial runner did not start")
570 }
571
572 const followUp = "I applied your review. What do you think now?"
573 acceptedAt := time.Now().Add(-time.Minute)
574 existing.setResumePrompt(followUp, acceptedAt)
575 f.persistAgent(existing)
576 if err := f.Stop(existing.ID); err != nil {
577 t.Fatal(err)
578 }
579 existing.Wait()
580
581 restarted, err := f.RestartTask(context.Background(), existing.ID)
582 if err != nil {
583 t.Fatal(err)
584 }
585 select {
586 case got := <-started:
587 if got != restarted {
588 t.Fatalf("restarted runner agent = %p, want %p", got, restarted)
589 }
590 resumePrompt, resumePromptAt := got.ResumePromptInfo()
591 if got.Resuming || resumePrompt != followUp || !resumePromptAt.Equal(acceptedAt) {
592 t.Fatalf("restarted lifecycle = resuming %t prompt %q accepted %s, want false, prompt %q, and accepted %s", got.Resuming, resumePrompt, resumePromptAt, followUp, acceptedAt)
593 }
594 case <-time.After(time.Second):
595 t.Fatal("restarted runner did not start")
596 }
597
598 persisted, err := readAgentMeta(filepath.Join(root, "agents", existing.ID))
599 if err != nil {
600 t.Fatal(err)
601 }
602 if persisted.ResumePrompt != followUp || !persisted.ResumePromptAt.Equal(acceptedAt) {
603 t.Fatalf("persisted restart prompt = %q accepted %s, want prompt %q and accepted %s", persisted.ResumePrompt, persisted.ResumePromptAt, followUp, acceptedAt)
604 }
605}
606
607func TestResumeRecalculatesInvalidStateInboxPath(t *testing.T) {
608 if runtime.GOOS == "windows" {
609 t.Skip("subagent inbox runtime relocation applies to Unix sockets")
610 }
611 root := t.TempDir()
612 f := New(Config{
613 Root: root, RepoRoot: root,
614 NewRunner: func(*Agent) Runner {
615 return RunnerFunc(func(context.Context, Sink) error { return nil })
616 },
617 })
618 a, err := f.Spawn(context.Background(), "invalid inbox")
619 if err != nil {
620 t.Fatal(err)
621 }
622 a.Wait()
623
624 invalidPath := filepath.Join(root, "agents", a.ID, "in.sock")
625 a.InboxPath = invalidPath
626 a2, err := f.Resume(context.Background(), a.ID)
627 if err != nil {
628 t.Fatalf("resume: %v", err)
629 }
630 a2.Wait()
631 if a2.InboxPath == invalidPath {
632 t.Fatalf("resume kept state-directory inbox path %q", invalidPath)
633 }
634 if strings.HasPrefix(a2.InboxPath, root+string(filepath.Separator)) {
635 t.Fatalf("resumed inbox %q remains below state root %q", a2.InboxPath, root)
636 }
637}
638
639// TestResumeSetsResumingFlag pins the contract execRunner relies on:
640// after Resume, the new Agent has Resuming == true; after a fresh
641// Spawn it's false. The runner branches on this to decide whether
642// to pass the original Task as a positional argv to the child, so
643// getting it wrong silently re-runs the initial task on every
644// resume and surfaces "agent busy; send 'cancel' first" between
645// assistant messages.
646func TestResumeSetsResumingFlag(t *testing.T) {
647 root := t.TempDir()
648 f := New(Config{
649 Root: root, RepoRoot: root,
650 NewRunner: func(a *Agent) Runner {
651 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
652 },
653 })
654 defer f.StopAll()
655
656 a, err := f.Spawn(context.Background(), "x")
657 if err != nil {
658 t.Fatal(err)
659 }
660 if a.Resuming {
661 t.Fatal("fresh Spawn produced Resuming==true; want false")
662 }
663 if err := f.Stop(a.ID); err != nil {
664 t.Fatal(err)
665 }
666 a.Wait()
667
668 a2, err := f.Resume(context.Background(), a.ID)
669 if err != nil {
670 t.Fatalf("resume: %v", err)
671 }
672 if !a2.Resuming {
673 t.Error("Resume produced Resuming==false; want true so the runner skips the duplicate initial-task argv")
674 }
675}
676
677func TestReloadClearsResumePromptAcknowledgedInEventLog(t *testing.T) {
678 root := t.TempDir()
679 stateDir := filepath.Join(root, "agents", "acknowledged-follow-up")
680 if err := os.MkdirAll(stateDir, 0o700); err != nil {
681 t.Fatal(err)
682 }
683 m := agentMeta{
684 ID: "acknowledged-follow-up",
685 Task: "review the implementation",
686 Dir: root,
687 Started: time.Now().Add(-time.Minute),
688 Status: StatusPending,
689 InboxPath: filepath.Join(stateDir, "in.sock"),
690 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
691 SessionPath: filepath.Join(stateDir, "session.json"),
692 ResumePrompt: "I applied your review. What do you think now?",
693 ResumePromptAt: time.Now().Add(-time.Second),
694 }
695 data, err := json.Marshal(m)
696 if err != nil {
697 t.Fatal(err)
698 }
699 if err := os.WriteFile(metaPath(stateDir), data, 0o600); err != nil {
700 t.Fatal(err)
701 }
702 log, err := OpenEventLog(m.EventLogPath)
703 if err != nil {
704 t.Fatal(err)
705 }
706 if err := log.Append(NewEvent("user_message", map[string]any{"content": []any{map[string]any{"type": "text", "text": m.ResumePrompt}}})); err != nil {
707 t.Fatal(err)
708 }
709 if err := log.Append(NewEvent(EventTurnStarted, map[string]any{"step": float64(1)})); err != nil {
710 t.Fatal(err)
711 }
712 if err := log.Close(); err != nil {
713 t.Fatal(err)
714 }
715
716 started := make(chan *Agent, 1)
717 f := New(Config{
718 Root: root, RepoRoot: root,
719 NewRunner: func(a *Agent) Runner {
720 return RunnerFunc(func(ctx context.Context, _ Sink) error {
721 started <- a
722 <-ctx.Done()
723 return ctx.Err()
724 })
725 },
726 })
727 t.Cleanup(f.StopAll)
728 if loaded, errs := f.Reload(); loaded != 1 || len(errs) != 0 {
729 t.Fatalf("reload = (%d, %v), want (1, no errors)", loaded, errs)
730 }
731 resumed, err := f.ResumeSession(context.Background(), m.ID)
732 if err != nil {
733 t.Fatal(err)
734 }
735 select {
736 case got := <-started:
737 if got != resumed || got.resumePrompt() != "" {
738 t.Fatalf("acknowledged follow-up was replayed as %q", got.resumePrompt())
739 }
740 case <-time.After(time.Second):
741 t.Fatal("resumed runner did not start")
742 }
743}
744
745func TestReloadRetainsQueuedResumePromptUntilTurnStarts(t *testing.T) {
746 root := t.TempDir()
747 const (
748 id = "queued-follow-up"
749 original = "review the implementation"
750 followUp = "I applied your review. What do you think now?"
751 )
752 stateDir := filepath.Join(root, "agents", id)
753 if err := os.MkdirAll(stateDir, 0o700); err != nil {
754 t.Fatal(err)
755 }
756 m := agentMeta{
757 ID: id,
758 Task: original,
759 OriginalTask: original,
760 Dir: root,
761 Started: time.Now().Add(-time.Minute),
762 Status: StatusPending,
763 ProcessState: ProcessPending,
764 TurnState: TurnQueued,
765 InboxPath: filepath.Join(stateDir, "in.sock"),
766 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
767 SessionPath: filepath.Join(stateDir, "session.json"),
768 ResumePrompt: followUp,
769 ResumePromptAt: time.Now(),
770 }
771 data, err := json.Marshal(m)
772 if err != nil {
773 t.Fatal(err)
774 }
775 if err := os.WriteFile(metaPath(stateDir), data, 0o600); err != nil {
776 t.Fatal(err)
777 }
778 if err := os.WriteFile(m.EventLogPath, nil, 0o600); err != nil {
779 t.Fatal(err)
780 }
781
782 started := make(chan *Agent, 1)
783 f := New(Config{
784 Root: root, RepoRoot: root,
785 NewRunner: func(a *Agent) Runner {
786 return RunnerFunc(func(ctx context.Context, _ Sink) error {
787 started <- a
788 <-ctx.Done()
789 return ctx.Err()
790 })
791 },
792 })
793 t.Cleanup(f.StopAll)
794 if loaded, errs := f.Reload(); loaded != 1 || len(errs) != 0 {
795 t.Fatalf("reload = (%d, %v), want (1, no errors)", loaded, errs)
796 }
797 loaded := f.Get(id)
798 if loaded == nil {
799 t.Fatal("reloaded agent missing")
800 }
801 if got := loaded.resumePrompt(); got != followUp {
802 t.Fatalf("reloaded pending prompt = %q, want %q", got, followUp)
803 }
804
805 resumed, err := f.ResumeSession(context.Background(), id)
806 if err != nil {
807 t.Fatal(err)
808 }
809 select {
810 case got := <-started:
811 if got != resumed || got.resumePrompt() != followUp {
812 t.Fatalf("resumed runner = %p prompt %q, want %p prompt %q", got, got.resumePrompt(), resumed, followUp)
813 }
814 case <-time.After(time.Second):
815 t.Fatal("resumed runner did not start")
816 }
817
818 updateAgentFromEvent(resumed, NewEvent(EventTurnStarted, map[string]any{"step": float64(1)}))
819 persisted, err := readAgentMeta(stateDir)
820 if err != nil {
821 t.Fatal(err)
822 }
823 if resumed.resumePrompt() != "" || persisted.ResumePrompt != "" {
824 t.Fatalf("pending prompt not cleared after turn start: agent %q meta %q", resumed.resumePrompt(), persisted.ResumePrompt)
825 }
826}
827
828func TestResumeWithPromptRetainsSessionAndStartsFollowUp(t *testing.T) {
829 root := t.TempDir()
830 started := make(chan *Agent, 2)
831 f := New(Config{
832 Root: root, RepoRoot: root,
833 NewRunner: func(a *Agent) Runner {
834 return RunnerFunc(func(ctx context.Context, _ Sink) error {
835 started <- a
836 <-ctx.Done()
837 return ctx.Err()
838 })
839 },
840 })
841 defer f.StopAll()
842
843 first, err := f.Spawn(context.Background(), "review the patch")
844 if err != nil {
845 t.Fatal(err)
846 }
847 select {
848 case got := <-started:
849 if got != first {
850 t.Fatalf("initial runner agent = %p, want %p", got, first)
851 }
852 case <-time.After(time.Second):
853 t.Fatal("initial runner did not start")
854 }
855 if err := f.Stop(first.ID); err != nil {
856 t.Fatal(err)
857 }
858 first.Wait()
859
860 const followUp = "I applied your feedback. Please review it again."
861 resumed, err := f.ResumeWithPrompt(context.Background(), first.ID, followUp)
862 if err != nil {
863 t.Fatalf("resume with prompt: %v", err)
864 }
865 if resumed.ID != first.ID || resumed.SessionPath != first.SessionPath {
866 t.Fatalf("resumed agent = id %q session %q, want id %q session %q", resumed.ID, resumed.SessionPath, first.ID, first.SessionPath)
867 }
868 resumePrompt, resumePromptAt := resumed.ResumePromptInfo()
869 if !resumed.Resuming || resumePrompt != followUp || resumePromptAt.IsZero() {
870 t.Fatalf("resumed lifecycle = resuming %t prompt %q accepted %s, want true, prompt %q, and timestamp", resumed.Resuming, resumePrompt, resumePromptAt, followUp)
871 }
872 persisted, err := readAgentMeta(filepath.Join(root, "agents", first.ID))
873 if err != nil {
874 t.Fatal(err)
875 }
876 if persisted.ResumePrompt != followUp || persisted.ResumePromptAt.IsZero() {
877 t.Fatalf("persisted follow-up = prompt %q accepted %s, want prompt %q and timestamp", persisted.ResumePrompt, persisted.ResumePromptAt, followUp)
878 }
879 select {
880 case got := <-started:
881 if got != resumed {
882 t.Fatalf("follow-up runner agent = %p, want %p", got, resumed)
883 }
884 resumePrompt, _ := got.ResumePromptInfo()
885 if resumePrompt != followUp {
886 t.Fatalf("runner follow-up = %q, want %q", resumePrompt, followUp)
887 }
888 case <-time.After(time.Second):
889 t.Fatal("resumed runner did not start")
890 }
891}
892
893// TestResumeRejectsRunningAgent prevents the user from double-running
894// an agent: two runners on the same session.json would race.
895func TestResumeRejectsRunningAgent(t *testing.T) {
896 root := t.TempDir()
897 f := New(Config{
898 Root: root, RepoRoot: root,
899 NewRunner: func(a *Agent) Runner {
900 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
901 },
902 })
903 defer f.StopAll()
904 a, err := f.Spawn(context.Background(), "x")
905 if err != nil {
906 t.Fatal(err)
907 }
908 // Wait for the runner to actually start.
909 deadline := time.Now().Add(time.Second)
910 for a.Status() != StatusRunning && time.Now().Before(deadline) {
911 time.Sleep(2 * time.Millisecond)
912 }
913 if _, err := f.Resume(context.Background(), a.ID); err == nil {
914 t.Fatal("resume on running agent did not error")
915 }
916}
917
918// TestResumeAfterReload exercises the full lifecycle the user cares
919// about: spawn in process A, exit A, start process B, Reload, Resume.
920// The resumed agent must keep its id and start producing new output
921// against the same on-disk state.
922//
923// Note on transcript persistence: in production the runner is
924// execRunner, which writes every event to events.jsonl, and Reload's
925// replayEventsIntoAgent rebuilds the transcript from there. The fake
926// RunnerFunc in this test calls sink.Transcript directly (which only
927// touches in-memory state), so the "first run" line is *expected* to
928// be lost across the restart — the in-memory Agent went away. We
929// have TestReloadReplaysTranscriptFromEventLog covering the
930// log-replay path with a curated events.jsonl.
931func TestResumeAfterReload(t *testing.T) {
932 root := t.TempDir()
933 // Process A
934 a := New(Config{
935 Root: root, RepoRoot: root,
936 NewRunner: func(ag *Agent) Runner {
937 return RunnerFunc(func(ctx context.Context, sink Sink) error {
938 sink.Transcript("first run for " + ag.ID)
939 <-ctx.Done()
940 return ctx.Err()
941 })
942 },
943 })
944 ag, err := a.Spawn(context.Background(), "x")
945 if err != nil {
946 t.Fatal(err)
947 }
948 // Let the runner record its transcript line before we tear it down.
949 deadline := time.Now().Add(time.Second)
950 for len(ag.Transcript()) == 0 && time.Now().Before(deadline) {
951 time.Sleep(2 * time.Millisecond)
952 }
953 id := ag.ID
954 a.StopAll()
955 ag.Wait()
956
957 // Process B: a new Supervisor against the same root.
958 resumed := make(chan struct{}, 1)
959 b := New(Config{
960 Root: root, RepoRoot: root,
961 NewRunner: func(ag *Agent) Runner {
962 return RunnerFunc(func(ctx context.Context, sink Sink) error {
963 sink.Transcript("second run for " + ag.ID)
964 select {
965 case resumed <- struct{}{}:
966 default:
967 }
968 <-ctx.Done()
969 return ctx.Err()
970 })
971 },
972 })
973 if loaded, errs := b.Reload(); loaded != 1 || len(errs) > 0 {
974 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
975 }
976 a2, err := b.Resume(context.Background(), id)
977 if err != nil {
978 t.Fatalf("resume: %v", err)
979 }
980 defer b.StopAll()
981
982 select {
983 case <-resumed:
984 case <-time.After(time.Second):
985 t.Fatal("resume runner did not start")
986 }
987 if a2.ID != id {
988 t.Errorf("resume id = %s; want %s", a2.ID, id)
989 }
990
991 // Wait for the new transcript line to land.
992 deadline = time.Now().Add(time.Second)
993 for time.Now().Before(deadline) {
994 for _, l := range a2.Transcript() {
995 if l == "second run for "+id {
996 return
997 }
998 }
999 time.Sleep(5 * time.Millisecond)
1000 }
1001 t.Fatalf("resumed transcript missing fresh line: %v", a2.Transcript())
1002}
1003
1004func TestStopMustFinishBeforeResumeOrRemove(t *testing.T) {
1005 root := t.TempDir()
1006 started := make(chan struct{})
1007 release := make(chan struct{})
1008 f := New(Config{
1009 Root: root, RepoRoot: root,
1010 NewRunner: func(*Agent) Runner {
1011 return RunnerFunc(func(context.Context, Sink) error {
1012 close(started)
1013 <-release
1014 return nil
1015 })
1016 },
1017 })
1018 a, err := f.Spawn(context.Background(), "slow stop")
1019 if err != nil {
1020 t.Fatal(err)
1021 }
1022 <-started
1023 if err := f.Stop(a.ID); err != nil {
1024 t.Fatal(err)
1025 }
1026 if _, err := f.Resume(context.Background(), a.ID); err == nil || !strings.Contains(err.Error(), "finalizing") {
1027 t.Fatalf("immediate resume error = %v, want finalizing", err)
1028 }
1029 if err := f.Remove(a.ID); err == nil || !strings.Contains(err.Error(), "finalizing") {
1030 t.Fatalf("immediate remove error = %v, want finalizing", err)
1031 }
1032 close(release)
1033 a.Wait()
1034}
1035
1036func TestResumeFencesLiveReloadedWorker(t *testing.T) {
1037 if runtime.GOOS == "windows" {
1038 t.Skip("subagent inbox transport uses Unix-domain sockets")
1039 }
1040 root := t.TempDir()
1041 first := New(Config{
1042 Root: root, RepoRoot: root,
1043 NewRunner: func(*Agent) Runner {
1044 return RunnerFunc(func(context.Context, Sink) error { return nil })
1045 },
1046 })
1047 a, err := first.Spawn(context.Background(), "live worker")
1048 if err != nil {
1049 t.Fatal(err)
1050 }
1051 a.Wait()
1052 ln, err := Listen(a.InboxPath)
1053 if err != nil {
1054 t.Fatal(err)
1055 }
1056 defer ln.Close()
1057
1058 second := New(Config{Root: root, RepoRoot: root})
1059 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
1060 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1061 }
1062 if got := second.Get(a.ID).ProcessState(); got != ProcessAlive {
1063 t.Fatalf("reloaded process state = %s, want alive", got)
1064 }
1065 if _, err := second.Resume(context.Background(), a.ID); err == nil || !strings.Contains(err.Error(), "live worker") {
1066 t.Fatalf("resume error = %v, want live-worker fence", err)
1067 }
1068}
1069
1070func TestSupervisorProviderSettingsPersistAcrossReloadAndResume(t *testing.T) {
1071 root := t.TempDir()
1072 newSupervisor := func(baseURL string, insecureTLS bool) *Supervisor {
1073 return New(Config{
1074 Root: root,
1075 RepoRoot: root,
1076 BaseURL: baseURL,
1077 InsecureTLS: insecureTLS,
1078 NewRunner: func(a *Agent) Runner {
1079 return RunnerFunc(func(ctx context.Context, _ Sink) error {
1080 <-ctx.Done()
1081 return ctx.Err()
1082 })
1083 },
1084 })
1085 }
1086
1087 first := newSupervisor("https://parent.example.test/v1", true)
1088 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "x"})
1089 if err != nil {
1090 t.Fatalf("spawn: %v", err)
1091 }
1092 if a.BaseURL != "https://parent.example.test/v1" || !a.InsecureTLS {
1093 t.Fatalf("spawned settings = (%q, %v), want inherited config", a.BaseURL, a.InsecureTLS)
1094 }
1095 if snap := a.Snapshot(); snap.BaseURL != a.BaseURL || snap.InsecureTLS != a.InsecureTLS {
1096 t.Fatalf("snapshot settings = (%q, %v), want (%q, %v)", snap.BaseURL, snap.InsecureTLS, a.BaseURL, a.InsecureTLS)
1097 }
1098 if err := first.Stop(a.ID); err != nil {
1099 t.Fatalf("stop: %v", err)
1100 }
1101 a.Wait()
1102
1103 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
1104 if err != nil {
1105 t.Fatalf("read meta.json: %v", err)
1106 }
1107 var meta agentMeta
1108 if err := json.Unmarshal(metaBytes, &meta); err != nil {
1109 t.Fatalf("decode meta.json: %v", err)
1110 }
1111 if meta.BaseURL != a.BaseURL || !meta.InsecureTLS {
1112 t.Fatalf("metadata settings = (%q, %v), want (%q, %v)", meta.BaseURL, meta.InsecureTLS, a.BaseURL, a.InsecureTLS)
1113 }
1114 if strings.Contains(string(metaBytes), "api_key") || strings.Contains(string(metaBytes), "parent-secret") {
1115 t.Fatalf("metadata contains credential material: %s", metaBytes)
1116 }
1117
1118 second := newSupervisor("https://different.example.test/v1", false)
1119 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
1120 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1121 }
1122 reloaded := second.Get(a.ID)
1123 if reloaded == nil || reloaded.BaseURL != a.BaseURL || !reloaded.InsecureTLS {
1124 t.Fatalf("reloaded settings = (%q, %v), want (%q, %v)", reloaded.BaseURL, reloaded.InsecureTLS, a.BaseURL, a.InsecureTLS)
1125 }
1126 resumed, err := second.Resume(context.Background(), a.ID)
1127 if err != nil {
1128 t.Fatalf("resume: %v", err)
1129 }
1130 if resumed.BaseURL != a.BaseURL || !resumed.InsecureTLS {
1131 t.Fatalf("resumed settings = (%q, %v), want (%q, %v)", resumed.BaseURL, resumed.InsecureTLS, a.BaseURL, a.InsecureTLS)
1132 }
1133 if err := second.Stop(resumed.ID); err != nil {
1134 t.Fatalf("stop resumed: %v", err)
1135 }
1136 resumed.Wait()
1137}
1138
1139func TestExplicitChildProviderDoesNotInheritHostEndpoint(t *testing.T) {
1140 f := New(Config{
1141 Root: t.TempDir(), RepoRoot: t.TempDir(), Provider: "anthropic",
1142 BaseURL: "https://host.example.test/v1", InsecureTLS: true,
1143 NewRunner: func(*Agent) Runner {
1144 return RunnerFunc(func(context.Context, Sink) error { return nil })
1145 },
1146 })
1147 a, err := f.SpawnReq(context.Background(), SpawnRequest{Task: "x", Provider: "openai"})
1148 if err != nil {
1149 t.Fatal(err)
1150 }
1151 a.Wait()
1152 if a.BaseURL != "" || a.InsecureTLS {
1153 t.Fatalf("explicit provider inherited host settings: (%q, %v)", a.BaseURL, a.InsecureTLS)
1154 }
1155}
1156
1157func TestSpawnRequestProviderSettingsOverrideConfig(t *testing.T) {
1158 root := t.TempDir()
1159 f := New(Config{
1160 Root: root,
1161 RepoRoot: root,
1162 BaseURL: "https://config.example.test/v1",
1163 NewRunner: func(*Agent) Runner {
1164 return RunnerFunc(func(context.Context, Sink) error { return nil })
1165 },
1166 })
1167 a, err := f.SpawnReq(context.Background(), SpawnRequest{
1168 Task: "x",
1169 BaseURL: "https://request.example.test/v1",
1170 InsecureTLS: true,
1171 })
1172 if err != nil {
1173 t.Fatalf("spawn: %v", err)
1174 }
1175 a.Wait()
1176 if a.BaseURL != "https://request.example.test/v1" || !a.InsecureTLS {
1177 t.Fatalf("spawned settings = (%q, %v), want request override", a.BaseURL, a.InsecureTLS)
1178 }
1179}
1180
1181// TestSpawnReqPersistsModel confirms that the per-agent model
1182// override is captured at Spawn time, surfaced via Snapshot, and
1183// written to meta.json so a later Reload + Resume reuses it.
1184func TestSpawnReqPersistsModel(t *testing.T) {
1185 root := t.TempDir()
1186 f := New(Config{
1187 Root: root, RepoRoot: root,
1188 NewRunner: func(a *Agent) Runner {
1189 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1190 },
1191 })
1192 f.SetActiveSession("host-session")
1193 a, err := f.SpawnReq(context.Background(), SpawnRequest{
1194 Task: "x", Model: "claude-sonnet-4-5", Provider: "anthropic",
1195 Reasoning: "high", Subagent: "reviewer",
1196 })
1197 if err != nil {
1198 t.Fatalf("spawn: %v", err)
1199 }
1200 if a.Model != "claude-sonnet-4-5" || a.Provider != "anthropic" || a.Reasoning != "high" || a.Subagent != "reviewer" || a.SessionID != "host-session" {
1201 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)
1202 }
1203 snap := a.Snapshot()
1204 if snap.Model != "claude-sonnet-4-5" || snap.Provider != "anthropic" || snap.Reasoning != "high" || snap.Subagent != "reviewer" {
1205 t.Fatalf("snapshot = (%q,%q,%q,%q); child fields not surfaced", snap.Model, snap.Provider, snap.Reasoning, snap.Subagent)
1206 }
1207
1208 // Stop so we can read meta.json without racing the run loop.
1209 if err := f.Stop(a.ID); err != nil {
1210 t.Fatal(err)
1211 }
1212 a.Wait()
1213
1214 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
1215 if err != nil {
1216 t.Fatalf("read meta.json: %v", err)
1217 }
1218 var got agentMeta
1219 if err := json.Unmarshal(metaBytes, &got); err != nil {
1220 t.Fatal(err)
1221 }
1222 if got.Model != "claude-sonnet-4-5" || got.Provider != "anthropic" || got.Reasoning != "high" || got.Subagent != "reviewer" || got.SessionID != "host-session" {
1223 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)
1224 }
1225
1226 // Reload in a fresh Supervisor and confirm the detached agent still
1227 // carries the model/provider so Resume can route the child
1228 // subprocess back to the same model.
1229 g := New(Config{
1230 Root: root, RepoRoot: root,
1231 NewRunner: func(a *Agent) Runner {
1232 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1233 },
1234 })
1235 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
1236 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1237 }
1238 re := g.Get(a.ID)
1239 if re == nil {
1240 t.Fatal("reloaded agent missing")
1241 }
1242 if re.Model != "claude-sonnet-4-5" || re.Provider != "anthropic" || re.Reasoning != "high" || re.Subagent != "reviewer" || re.SessionID != "host-session" {
1243 t.Errorf("reloaded fields = (%q,%q,%q,%q,%q); want preserved", re.Model, re.Provider, re.Reasoning, re.Subagent, re.SessionID)
1244 }
1245
1246 resumed, err := g.Resume(context.Background(), re.ID)
1247 if err != nil {
1248 t.Fatalf("resume: %v", err)
1249 }
1250 if resumed.Reasoning != "high" || resumed.Subagent != "reviewer" || resumed.SessionID != "host-session" {
1251 t.Errorf("resumed fields = (%q,%q,%q); want reasoning/profile/session preserved", resumed.Reasoning, resumed.Subagent, resumed.SessionID)
1252 }
1253 if err := g.Stop(resumed.ID); err != nil {
1254 t.Fatalf("stop resumed agent: %v", err)
1255 }
1256 resumed.Wait()
1257}
1258
1259// TestSubagentWorkerArgsIncludesModelFlags pins the argv contract that
1260// connects the supervisor's per-agent model override to the child's
1261// --model / --provider flag set. Adding a new path to argv without
1262// updating this assertion is the failure mode this catches.
1263func TestSubagentWorkerArgsIncludesModelFlags(t *testing.T) {
1264 args := subagentWorkerArgs(subagentWorkerArgsOpts{
1265 Exe: "/zut", Dir: "/wt", SessionPath: "/s.json", InboxPath: "/in.sock",
1266 Task: "do x", Model: "gpt-5", Provider: "openai",
1267 })
1268 want := []string{"--model", "gpt-5", "--provider", "openai"}
1269 for i := 0; i+1 < len(want); i += 2 {
1270 flag, value := want[i], want[i+1]
1271 found := false
1272 for j := 0; j+1 < len(args); j++ {
1273 if args[j] == flag && args[j+1] == value {
1274 found = true
1275 break
1276 }
1277 }
1278 if !found {
1279 t.Errorf("argv missing %s %s: %v", flag, value, args)
1280 }
1281 }
1282 // And the task still ends up positional last.
1283 if args[len(args)-1] != "do x" {
1284 t.Errorf("task should be last positional; argv=%v", args)
1285 }
1286}
1287
1288// TestStopOnDetachedAgentIsNoopAndDoesNotPanic regression-tests the
1289// segfault that hit when the user pressed 'k' on a reloaded
1290// (detached) agent: Supervisor.Stop unconditionally called a.cancel(),
1291// but buildDetachedAgent never assigns a cancel func because there's
1292// no in-process runner to cancel. The fix is to short-circuit Stop
1293// for StatusDetached and — belt-and-braces — nil-check a.cancel.
1294func TestStopOnDetachedAgentIsNoopAndDoesNotPanic(t *testing.T) {
1295 root := t.TempDir()
1296 // Build a detached agent the same way Reload does: drop a
1297 // meta.json on disk, then ask a fresh Supervisor to pick it up.
1298 id := "alpha-1"
1299 stateDir := filepath.Join(root, "agents", id)
1300 if err := os.MkdirAll(stateDir, 0o755); err != nil {
1301 t.Fatal(err)
1302 }
1303 m := agentMeta{
1304 ID: id, Task: "t",
1305 Dir: root,
1306 Started: time.Now().Add(-time.Hour),
1307 InboxPath: filepath.Join(stateDir, "in.sock"),
1308 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
1309 SessionPath: filepath.Join(stateDir, "session.json"),
1310 }
1311 mb, _ := json.MarshalIndent(m, "", " ")
1312 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
1313 t.Fatal(err)
1314 }
1315
1316 f := New(Config{Root: root, RepoRoot: root})
1317 if loaded, errs := f.Reload(); loaded != 1 || len(errs) > 0 {
1318 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1319 }
1320 a := f.Get(id)
1321 if a == nil {
1322 t.Fatal("detached agent missing from subagent")
1323 }
1324 if a.Status() != StatusDetached {
1325 t.Fatalf("setup: agent status = %q; want detached", a.Status())
1326 }
1327
1328 // The real assertion: Stop returns cleanly without panicking.
1329 defer func() {
1330 if r := recover(); r != nil {
1331 t.Fatalf("Stop on detached agent panicked: %v", r)
1332 }
1333 }()
1334 if err := f.Stop(id); err != nil {
1335 t.Fatalf("Stop on detached agent: %v", err)
1336 }
1337 // Status must remain Detached (Stop is a no-op).
1338 if got := a.Status(); got != StatusDetached {
1339 t.Errorf("after Stop status = %q; want detached (no-op)", got)
1340 }
1341
1342 // StopAll, which the cli defers on exit, must also be safe.
1343 f.StopAll()
1344}
1345
1346// TestRemoveAlsoCleansStateDir ensures Remove deletes the on-disk
1347// state directory in addition to the worktree, so a removed agent
1348// doesn't reappear on the next Reload.
1349func TestRemoveAlsoCleansStateDir(t *testing.T) {
1350 root := t.TempDir()
1351 f := New(Config{
1352 Root: root, RepoRoot: root,
1353 NewRunner: func(a *Agent) Runner {
1354 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1355 },
1356 })
1357 a, err := f.Spawn(context.Background(), "x")
1358 if err != nil {
1359 t.Fatal(err)
1360 }
1361 stateDir := filepath.Join(root, "agents", a.ID)
1362 if _, err := os.Stat(filepath.Join(stateDir, "meta.json")); err != nil {
1363 t.Fatalf("meta.json missing pre-remove: %v", err)
1364 }
1365 if err := f.Stop(a.ID); err != nil {
1366 t.Fatal(err)
1367 }
1368 a.Wait()
1369 if err := f.Remove(a.ID); err != nil {
1370 t.Fatal(err)
1371 }
1372 if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
1373 t.Fatalf("state dir should be gone after remove; got %v", err)
1374 }
1375
1376 // A fresh Supervisor + Reload should find nothing.
1377 g := New(Config{Root: root, RepoRoot: root})
1378 if loaded, _ := g.Reload(); loaded != 0 {
1379 t.Fatalf("reload after remove loaded=%d; want 0", loaded)
1380 }
1381}
1382
1383func TestRemoveReloadedAgentKeepsSiblingState(t *testing.T) {
1384 root := t.TempDir()
1385 first := New(Config{
1386 Root: root, RepoRoot: root,
1387 NewRunner: func(*Agent) Runner {
1388 return RunnerFunc(func(context.Context, Sink) error { return nil })
1389 },
1390 })
1391 a1, err := first.Spawn(context.Background(), "first")
1392 if err != nil {
1393 t.Fatal(err)
1394 }
1395 a2, err := first.Spawn(context.Background(), "second")
1396 if err != nil {
1397 t.Fatal(err)
1398 }
1399 a1.Wait()
1400 a2.Wait()
1401
1402 second := New(Config{Root: root, RepoRoot: root})
1403 if loaded, errs := second.Reload(); loaded != 2 || len(errs) != 0 {
1404 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
1405 }
1406 if err := second.Remove(a1.ID); err != nil {
1407 t.Fatal(err)
1408 }
1409 if _, err := os.Stat(filepath.Join(root, "agents", a2.ID, "meta.json")); err != nil {
1410 t.Fatalf("removing reloaded sibling deleted surviving metadata: %v", err)
1411 }
1412}
1413
1414// TestActiveSessionScopesSnapshotAll proves the session filter the
1415// user asked for: a single Supervisor with agents spawned under two
1416// different active sessions only surfaces the agents matching the
1417// currently-active session via SnapshotAll. Switching the active
1418// session re-narrows the view without touching agent state.
1419func TestActiveSessionScopesSnapshotAll(t *testing.T) {
1420 root := t.TempDir()
1421 f := New(Config{
1422 Root: root, RepoRoot: root,
1423 NewRunner: func(a *Agent) Runner {
1424 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1425 },
1426 })
1427
1428 // Spawn one agent under session A and one under session B.
1429 f.SetActiveSession("sess-A")
1430 aA, err := f.Spawn(context.Background(), "task A")
1431 if err != nil {
1432 t.Fatal(err)
1433 }
1434 f.SetActiveSession("sess-B")
1435 aB, err := f.Spawn(context.Background(), "task B")
1436 if err != nil {
1437 t.Fatal(err)
1438 }
1439
1440 // Both agents must carry the session id they were spawned under,
1441 // regardless of any later SetActiveSession.
1442 if aA.SessionID != "sess-A" {
1443 t.Errorf("aA.SessionID = %q; want sess-A", aA.SessionID)
1444 }
1445 if aB.SessionID != "sess-B" {
1446 t.Errorf("aB.SessionID = %q; want sess-B", aB.SessionID)
1447 }
1448
1449 // Active session B → only aB visible.
1450 only := snapshotIDs(f.SnapshotAll())
1451 if len(only) != 1 || only[0] != aB.ID {
1452 t.Errorf("scoped to sess-B, snapshot ids = %v; want [%s]", only, aB.ID)
1453 }
1454
1455 // Switch back to A → only aA visible.
1456 f.SetActiveSession("sess-A")
1457 only = snapshotIDs(f.SnapshotAll())
1458 if len(only) != 1 || only[0] != aA.ID {
1459 t.Errorf("scoped to sess-A, snapshot ids = %v; want [%s]", only, aA.ID)
1460 }
1461
1462 // Clear the scope → both visible.
1463 f.SetActiveSession("")
1464 all := snapshotIDs(f.SnapshotAll())
1465 if len(all) != 2 {
1466 t.Errorf("unscoped snapshot ids = %v; want both agents", all)
1467 }
1468
1469 // Cleanup.
1470 _ = f.Stop(aA.ID)
1471 _ = f.Stop(aB.ID)
1472 aA.Wait()
1473 aB.Wait()
1474}
1475
1476// TestSessionIDPersistsAcrossReload confirms the SessionID field
1477// survives a Supervisor restart: an agent spawned under session A in one
1478// Supervisor instance is still SessionID="sess-A" after a fresh New +
1479// Reload reads it back from meta.json. Without persistence, the
1480// scope filter would forget which session owned each agent after
1481// a zut restart and the dashboard would leak everything again.
1482func TestSessionIDPersistsAcrossReload(t *testing.T) {
1483 root := t.TempDir()
1484 mkSupervisor := func() *Supervisor {
1485 return New(Config{
1486 Root: root, RepoRoot: root,
1487 NewRunner: func(a *Agent) Runner {
1488 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1489 },
1490 })
1491 }
1492
1493 f := mkSupervisor()
1494 f.SetActiveSession("sess-keep")
1495 a, err := f.Spawn(context.Background(), "persist me")
1496 if err != nil {
1497 t.Fatal(err)
1498 }
1499 _ = f.Stop(a.ID)
1500 a.Wait()
1501
1502 // Fresh Supervisor, same root, reload from disk.
1503 g := mkSupervisor()
1504 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
1505 t.Fatalf("reload loaded=%d errs=%v; want 1 / no errs", loaded, errs)
1506 }
1507 g.SetActiveSession("sess-keep")
1508 got := snapshotIDs(g.SnapshotAll())
1509 if len(got) != 1 || got[0] != a.ID {
1510 t.Errorf("after reload + scope to sess-keep, ids = %v; want [%s]", got, a.ID)
1511 }
1512
1513 // A terminal persisted agent remains visible from another host session
1514 // so a normal restart does not hide durable recovery state.
1515 g.SetActiveSession("sess-other")
1516 if got := snapshotIDs(g.SnapshotAll()); len(got) != 1 || got[0] != a.ID {
1517 t.Errorf("scoped to other session, ids = %v; want [%s]", got, a.ID)
1518 }
1519}
1520
1521// TestEmptySessionIDIsVisibleFromAnyScope pins the unscoped-agent
1522// rule: agents whose meta.json predates the SessionID field (or who
1523// were spawned without an active session, e.g. via a test rig or
1524// scripted caller) carry SessionID == "" and remain visible from
1525// every scope. Otherwise the schema bump would orphan every
1526// pre-existing agent the moment a user upgraded zut.
1527func TestEmptySessionIDIsVisibleFromAnyScope(t *testing.T) {
1528 root := t.TempDir()
1529 f := New(Config{
1530 Root: root, RepoRoot: root,
1531 NewRunner: func(a *Agent) Runner {
1532 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
1533 },
1534 })
1535 // No SetActiveSession call → agent spawned with empty SessionID.
1536 a, err := f.Spawn(context.Background(), "unscoped")
1537 if err != nil {
1538 t.Fatal(err)
1539 }
1540 if a.SessionID != "" {
1541 t.Fatalf("unscoped spawn produced SessionID %q; want empty", a.SessionID)
1542 }
1543
1544 for _, scope := range []string{"", "any-session", "some-other"} {
1545 f.SetActiveSession(scope)
1546 got := snapshotIDs(f.SnapshotAll())
1547 if len(got) != 1 || got[0] != a.ID {
1548 t.Errorf("scope=%q: ids = %v; want unscoped agent visible", scope, got)
1549 }
1550 }
1551
1552 _ = f.Stop(a.ID)
1553 a.Wait()
1554}
1555
1556func TestWebSearchPolicyPersistsThroughReloadResumeAndChildArgv(t *testing.T) {
1557 root := t.TempDir()
1558 newSupervisor := func(policy WebSearchPolicy) *Supervisor {
1559 return New(Config{
1560 Root: root,
1561 RepoRoot: root,
1562 WebSearchPolicy: policy,
1563 NewRunner: func(*Agent) Runner {
1564 return RunnerFunc(func(context.Context, Sink) error { return nil })
1565 },
1566 })
1567 }
1568
1569 first := newSupervisor(WebSearchAllow)
1570 t.Cleanup(first.StopAll)
1571 spawned, err := first.Spawn(context.Background(), "web policy")
1572 if err != nil {
1573 t.Fatalf("spawn: %v", err)
1574 }
1575 spawned.Wait()
1576 if spawned.WebSearchPolicy != WebSearchAllow {
1577 t.Fatalf("spawned policy = %v, want allow", spawned.WebSearchPolicy)
1578 }
1579
1580 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", spawned.ID, "meta.json"))
1581 if err != nil {
1582 t.Fatalf("read metadata: %v", err)
1583 }
1584 var meta agentMeta
1585 if err := json.Unmarshal(metaBytes, &meta); err != nil {
1586 t.Fatalf("decode metadata: %v", err)
1587 }
1588 if meta.WebSearchPolicy != WebSearchAllow {
1589 t.Fatalf("metadata policy = %v, want allow", meta.WebSearchPolicy)
1590 }
1591
1592 second := newSupervisor(WebSearchDeny)
1593 t.Cleanup(second.StopAll)
1594 if loaded, errs := second.Reload(); loaded != 1 || len(errs) != 0 {
1595 t.Fatalf("reload loaded=%d errs=%v; want one clean load", loaded, errs)
1596 }
1597 reloaded := second.Get(spawned.ID)
1598 if reloaded == nil {
1599 t.Fatal("reloaded agent missing")
1600 }
1601 if reloaded.WebSearchPolicy != WebSearchAllow {
1602 t.Fatalf("reloaded policy = %v, want allow", reloaded.WebSearchPolicy)
1603 }
1604
1605 resumed, err := second.ResumeSession(context.Background(), spawned.ID)
1606 if err != nil {
1607 t.Fatalf("resume: %v", err)
1608 }
1609 if resumed.WebSearchPolicy != WebSearchAllow {
1610 t.Fatalf("resumed policy = %v, want allow", resumed.WebSearchPolicy)
1611 }
1612 args := defaultChildArgs("/zut", resumed, "/state/session.json", "/state/inbox.sock")
1613 idx := indexOf(args, "--web-search-policy")
1614 if idx < 0 || safeAt(args, idx+1) != "allow" {
1615 t.Fatalf("resumed child argv = %v, want explicit allow", args)
1616 }
1617 resumed.Wait()
1618}
1619
1620func snapshotIDs(ss []AgentSnapshot) []string {
1621 out := make([]string, len(ss))
1622 for i, s := range ss {
1623 out[i] = s.ID
1624 }
1625 return out
1626}