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