1package swarm
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 zot launch can't
20// find the agent and the user loses access to the worktree.
21func TestSpawnWritesMetaJSON(t *testing.T) {
22 root := t.TempDir()
23 f := New(Config{
24 Root: root,
25 RepoRoot: root,
26 NewRunner: func(a *Agent) Runner {
27 return RunnerFunc(func(ctx context.Context, _ Sink) error {
28 <-ctx.Done()
29 return ctx.Err()
30 })
31 },
32 })
33 defer f.StopAll()
34
35 a, err := f.Spawn(context.Background(), "investigate widget")
36 if err != nil {
37 t.Fatalf("spawn: %v", err)
38 }
39
40 // Read meta.json straight off disk.
41 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
42 if err != nil {
43 t.Fatalf("read meta.json: %v", err)
44 }
45 var got agentMeta
46 if err := json.Unmarshal(metaBytes, &got); err != nil {
47 t.Fatalf("parse meta.json: %v", err)
48 }
49 if got.ID != a.ID {
50 t.Errorf("meta.ID = %q; want %q", got.ID, a.ID)
51 }
52 if got.Task != "investigate widget" {
53 t.Errorf("meta.Task = %q", got.Task)
54 }
55 if got.Dir != a.Dir {
56 t.Errorf("meta paths drifted: %+v vs agent %+v", got, a)
57 }
58 if got.InboxPath == "" || got.EventLogPath == "" || got.SessionPath == "" {
59 t.Errorf("meta paths empty: %+v", got)
60 }
61 if got.Started.IsZero() {
62 t.Error("meta.Started is zero")
63 }
64}
65
66// TestReloadRebuildsDetachedAgents simulates a zot restart by spawning
67// in one Swarm, throwing it away, then opening a fresh Swarm against
68// the same root and calling Reload. The user-visible state — id,
69// task, branch, dir — must come back, and status must be Detached so
70// the dashboard can offer resume.
71func TestReloadRebuildsDetachedAgents(t *testing.T) {
72 root := t.TempDir()
73
74 // First incarnation: spawn two agents, then drop the supervisor.
75 first := New(Config{
76 Root: root,
77 RepoRoot: root,
78 NewRunner: func(a *Agent) Runner {
79 return RunnerFunc(func(ctx context.Context, _ Sink) error {
80 <-ctx.Done()
81 return ctx.Err()
82 })
83 },
84 })
85 a1, err := first.Spawn(context.Background(), "alpha task")
86 if err != nil {
87 t.Fatalf("spawn a1: %v", err)
88 }
89 a2, err := first.Spawn(context.Background(), "beta task")
90 if err != nil {
91 t.Fatalf("spawn a2: %v", err)
92 }
93 first.StopAll()
94 // Wait briefly for the runner goroutines to settle so their
95 // stop event reaches the log before we move on. Reload itself
96 // doesn't need this; it makes the assertions below deterministic.
97 a1.Wait()
98 a2.Wait()
99
100 // Second incarnation against the same root.
101 second := New(Config{
102 Root: root,
103 RepoRoot: root,
104 })
105 loaded, errs := second.Reload()
106 if len(errs) > 0 {
107 t.Fatalf("reload errs: %v", errs)
108 }
109 if loaded != 2 {
110 t.Fatalf("loaded = %d; want 2", loaded)
111 }
112 snap := second.SnapshotAll()
113 if len(snap) != 2 {
114 t.Fatalf("snapshot len = %d; want 2", len(snap))
115 }
116 ids := map[string]AgentSnapshot{}
117 for _, s := range snap {
118 ids[s.ID] = s
119 }
120 if _, ok := ids[a1.ID]; !ok {
121 t.Errorf("reloaded set missing %s", a1.ID)
122 }
123 if _, ok := ids[a2.ID]; !ok {
124 t.Errorf("reloaded set missing %s", a2.ID)
125 }
126 for _, s := range snap {
127 if s.Status != StatusDetached {
128 t.Errorf("agent %s status = %q; want detached", s.ID, s.Status)
129 }
130 if s.Task == "" {
131 t.Errorf("agent %s lost its task", s.ID)
132 }
133 }
134}
135
136// TestReloadIsIdempotent calls Reload twice in a row and asserts the
137// second call neither duplicates rows nor errors.
138func TestReloadIsIdempotent(t *testing.T) {
139 root := t.TempDir()
140 first := New(Config{
141 Root: root, RepoRoot: root,
142 NewRunner: func(a *Agent) Runner {
143 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
144 },
145 })
146 if _, err := first.Spawn(context.Background(), "only one"); err != nil {
147 t.Fatal(err)
148 }
149 first.StopAll()
150
151 second := New(Config{Root: root, RepoRoot: root})
152 loaded1, _ := second.Reload()
153 loaded2, errs := second.Reload()
154 if len(errs) > 0 {
155 t.Fatalf("errs on second reload: %v", errs)
156 }
157 if loaded1 != 1 || loaded2 != 0 {
158 t.Fatalf("loaded counts = (%d, %d); want (1, 0)", loaded1, loaded2)
159 }
160 if got := len(second.SnapshotAll()); got != 1 {
161 t.Fatalf("snapshot len = %d; want 1", got)
162 }
163}
164
165// TestReloadReplaysTranscriptFromEventLog drops a curated events.jsonl
166// next to a meta.json and checks that the transcript surfaces in the
167// reloaded agent. This is the user-facing payoff of Reload: opening
168// /swarm logs <id> after a restart shows what the agent said before.
169func TestReloadReplaysTranscriptFromEventLog(t *testing.T) {
170 root := t.TempDir()
171 id := "alpha-9"
172 stateDir := filepath.Join(root, "agents", id)
173 if err := os.MkdirAll(stateDir, 0o755); err != nil {
174 t.Fatal(err)
175 }
176 // meta.json
177 m := agentMeta{
178 ID: id, Task: "do thing",
179 Dir: root, Started: time.Now().Add(-time.Hour),
180 InboxPath: filepath.Join(stateDir, "in.sock"),
181 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
182 SessionPath: filepath.Join(stateDir, "session.json"),
183 }
184 mb, _ := json.MarshalIndent(m, "", " ")
185 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
186 t.Fatal(err)
187 }
188 // Event log with one assistant turn and a graceful stop.
189 log, err := OpenEventLog(m.EventLogPath)
190 if err != nil {
191 t.Fatal(err)
192 }
193 _ = log.Append(NewEvent("assistant_message", map[string]any{
194 "content": []any{map[string]any{"type": "text", "text": "hello from the past"}},
195 }))
196 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "shutdown"}))
197 _ = log.Close()
198
199 f := New(Config{Root: root, RepoRoot: root})
200 loaded, errs := f.Reload()
201 if len(errs) > 0 || loaded != 1 {
202 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
203 }
204 a := f.Get(id)
205 if a == nil {
206 t.Fatal("reloaded agent not found")
207 }
208 // Status should reflect the lifecycle terminator we wrote.
209 if a.Status() != StatusDone {
210 t.Errorf("status = %q; want done (offline)", a.Status())
211 }
212 got := a.Transcript()
213 found := false
214 for _, line := range got {
215 if line == "hello from the past" {
216 found = true
217 }
218 }
219 if !found {
220 t.Errorf("transcript missing replayed line: %v", got)
221 }
222}
223
224func TestDetachedReplayKeepsLastAnswerAcrossLargeEvents(t *testing.T) {
225 root := t.TempDir()
226 path := filepath.Join(root, "events.jsonl")
227 log, err := OpenEventLog(path)
228 if err != nil {
229 t.Fatal(err)
230 }
231 for _, ev := range []Event{
232 NewEvent("assistant_message", map[string]any{
233 "content": []any{map[string]any{"type": "text", "text": "final answer"}},
234 }),
235 NewEvent("tool_result", map[string]any{"data": strings.Repeat("x", detachedReplayBytes)}),
236 NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}),
237 } {
238 if err := log.Append(ev); err != nil {
239 t.Fatal(err)
240 }
241 }
242 if err := log.Close(); err != nil {
243 t.Fatal(err)
244 }
245
246 f := New(Config{Root: root, RepoRoot: root})
247 a := f.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path})
248 if got := a.Snapshot(); got.LastAssistant != "" || len(got.Lines) != 0 || got.Status != StatusDone {
249 t.Fatalf("startup should use a bounded tail: assistant=%q, lines=%v, status=%s", got.LastAssistant, got.Lines, got.Status)
250 }
251 f.agents[a.ID] = a
252 f.order = append(f.order, a.ID)
253 got := f.SnapshotAll()
254 if len(got) != 1 || got[0].LastAssistant != "final answer" ||
255 len(got[0].Lines) != 1 || got[0].Lines[0] != "final answer" || got[0].Status != StatusDone {
256 t.Fatalf("dashboard snapshot: %v", got)
257 }
258
259 // Resume must backfill even when the dashboard was never opened.
260 f2 := New(Config{Root: root, RepoRoot: root, NewRunner: func(*Agent) Runner {
261 return RunnerFunc(func(ctx context.Context, _ Sink) error {
262 <-ctx.Done()
263 return ctx.Err()
264 })
265 }})
266 defer f2.StopAll()
267 b := f2.buildDetachedAgent(agentMeta{ID: "large", EventLogPath: path})
268 f2.agents[b.ID] = b
269 f2.order = append(f2.order, b.ID)
270 resumed, err := f2.Resume(context.Background(), b.ID)
271 if err != nil {
272 t.Fatal(err)
273 }
274 if lines := resumed.Transcript(); len(lines) != 1 || lines[0] != "final answer" {
275 t.Fatalf("resumed transcript: %v", lines)
276 }
277}
278
279// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
280// doesn't blow up the whole reload. Directories with no meta.json
281// at all are silently ignored (Spawn that failed mid-way leaves
282// these); corrupt meta files are reported as errors but don't stop
283// the rest of the load.
284func TestReloadSkipsBareDirsAndCorruptMeta(t *testing.T) {
285 root := t.TempDir()
286 agentsDir := filepath.Join(root, "agents")
287 // Bare directory, no meta.json.
288 if err := os.MkdirAll(filepath.Join(agentsDir, "leftover"), 0o755); err != nil {
289 t.Fatal(err)
290 }
291 // Directory with garbage meta.json.
292 if err := os.MkdirAll(filepath.Join(agentsDir, "corrupt"), 0o755); err != nil {
293 t.Fatal(err)
294 }
295 if err := os.WriteFile(filepath.Join(agentsDir, "corrupt", "meta.json"), []byte("not json"), 0o644); err != nil {
296 t.Fatal(err)
297 }
298 // One good agent.
299 good := "good-1"
300 stateDir := filepath.Join(agentsDir, good)
301 _ = os.MkdirAll(stateDir, 0o755)
302 m := agentMeta{ID: good, Task: "x", Dir: "/tmp/x", Started: time.Now()}
303 mb, _ := json.MarshalIndent(m, "", " ")
304 _ = os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644)
305
306 f := New(Config{Root: root, RepoRoot: root})
307 loaded, errs := f.Reload()
308 if loaded != 1 {
309 t.Errorf("loaded = %d; want 1", loaded)
310 }
311 if len(errs) != 1 {
312 t.Errorf("errs len = %d; want 1 (the corrupt entry): %v", len(errs), errs)
313 }
314}
315
316// TestResumeRestartsRunnerOnSameSession is the headline test. We:
317//
318// 1. Spawn an agent with a counting runner (records each Run call).
319// 2. Stop it.
320// 3. Resume it.
321// 4. Assert Run was called twice — once for spawn, once for resume —
322// against the same SessionPath / InboxPath / Dir.
323func TestResumeRestartsRunnerOnSameSession(t *testing.T) {
324 root := t.TempDir()
325 var (
326 mu sync.Mutex
327 runs []*Agent
328 runsDone int32
329 )
330 f := New(Config{
331 Root: root, RepoRoot: root,
332 NewRunner: func(a *Agent) Runner {
333 return RunnerFunc(func(ctx context.Context, sink Sink) error {
334 mu.Lock()
335 runs = append(runs, a)
336 mu.Unlock()
337 sink.Activity("ran")
338 atomic.AddInt32(&runsDone, 1)
339 <-ctx.Done()
340 return ctx.Err()
341 })
342 },
343 })
344 defer f.StopAll()
345
346 a, err := f.Spawn(context.Background(), "do thing")
347 if err != nil {
348 t.Fatalf("spawn: %v", err)
349 }
350 originalSession := a.SessionPath
351 originalDir := a.Dir
352 originalInbox := a.InboxPath
353
354 // Wait for the runner to record its presence, then stop the agent.
355 deadline := time.Now().Add(time.Second)
356 for atomic.LoadInt32(&runsDone) < 1 && time.Now().Before(deadline) {
357 time.Sleep(2 * time.Millisecond)
358 }
359 if err := f.Stop(a.ID); err != nil {
360 t.Fatalf("stop: %v", err)
361 }
362 a.Wait()
363 if a.Status() != StatusKilled {
364 t.Fatalf("post-stop status = %q; want killed", a.Status())
365 }
366
367 // Now resume.
368 a2, err := f.Resume(context.Background(), a.ID)
369 if err != nil {
370 t.Fatalf("resume: %v", err)
371 }
372 if a2.ID != a.ID {
373 t.Errorf("resume returned different id: %s vs %s", a2.ID, a.ID)
374 }
375 if a2.SessionPath != originalSession {
376 t.Errorf("resume changed session path: %s vs %s", a2.SessionPath, originalSession)
377 }
378 if a2.Dir != originalDir {
379 t.Errorf("resume changed dir: %s vs %s", a2.Dir, originalDir)
380 }
381 if a2.InboxPath != originalInbox {
382 t.Errorf("resume changed inbox path: %s vs %s", a2.InboxPath, originalInbox)
383 }
384 // Two runner invocations: spawn + resume.
385 deadline = time.Now().Add(time.Second)
386 for atomic.LoadInt32(&runsDone) < 2 && time.Now().Before(deadline) {
387 time.Sleep(2 * time.Millisecond)
388 }
389 mu.Lock()
390 defer mu.Unlock()
391 if len(runs) != 2 {
392 t.Fatalf("runner invocations = %d; want 2", len(runs))
393 }
394 if runs[1].ID != a.ID {
395 t.Errorf("second run had wrong id: %s", runs[1].ID)
396 }
397 if got := len(f.SnapshotAll()); got != 1 {
398 t.Errorf("snapshot len = %d; want 1 (in-place replace)", got)
399 }
400}
401
402func TestResumeRecalculatesLegacyStateInboxPath(t *testing.T) {
403 if runtime.GOOS == "windows" {
404 t.Skip("swarm inbox runtime relocation applies to Unix sockets")
405 }
406 root := t.TempDir()
407 f := New(Config{
408 Root: root, RepoRoot: root,
409 NewRunner: func(*Agent) Runner {
410 return RunnerFunc(func(context.Context, Sink) error { return nil })
411 },
412 })
413 a, err := f.Spawn(context.Background(), "legacy inbox")
414 if err != nil {
415 t.Fatal(err)
416 }
417 a.Wait()
418
419 legacyPath := filepath.Join(root, "agents", a.ID, "in.sock")
420 a.InboxPath = legacyPath
421 a2, err := f.Resume(context.Background(), a.ID)
422 if err != nil {
423 t.Fatalf("resume: %v", err)
424 }
425 a2.Wait()
426 if a2.InboxPath == legacyPath {
427 t.Fatalf("resume kept legacy state-directory inbox path %q", legacyPath)
428 }
429 if strings.HasPrefix(a2.InboxPath, root+string(filepath.Separator)) {
430 t.Fatalf("resumed inbox %q remains below state root %q", a2.InboxPath, root)
431 }
432}
433
434// TestResumeSetsResumingFlag pins the contract execRunner relies on:
435// after Resume, the new Agent has Resuming == true; after a fresh
436// Spawn it's false. The runner branches on this to decide whether
437// to pass the original Task as a positional argv to the child, so
438// getting it wrong silently re-runs the initial task on every
439// resume and surfaces "agent busy; send 'cancel' first" between
440// assistant messages.
441func TestResumeSetsResumingFlag(t *testing.T) {
442 root := t.TempDir()
443 f := New(Config{
444 Root: root, RepoRoot: root,
445 NewRunner: func(a *Agent) Runner {
446 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
447 },
448 })
449 defer f.StopAll()
450
451 a, err := f.Spawn(context.Background(), "x")
452 if err != nil {
453 t.Fatal(err)
454 }
455 if a.Resuming {
456 t.Fatal("fresh Spawn produced Resuming==true; want false")
457 }
458 if err := f.Stop(a.ID); err != nil {
459 t.Fatal(err)
460 }
461 a.Wait()
462
463 a2, err := f.Resume(context.Background(), a.ID)
464 if err != nil {
465 t.Fatalf("resume: %v", err)
466 }
467 if !a2.Resuming {
468 t.Error("Resume produced Resuming==false; want true so the runner skips the duplicate initial-task argv")
469 }
470}
471
472// TestResumeRejectsRunningAgent prevents the user from double-running
473// an agent: two runners on the same session.json would race.
474func TestResumeRejectsRunningAgent(t *testing.T) {
475 root := t.TempDir()
476 f := New(Config{
477 Root: root, RepoRoot: root,
478 NewRunner: func(a *Agent) Runner {
479 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
480 },
481 })
482 defer f.StopAll()
483 a, err := f.Spawn(context.Background(), "x")
484 if err != nil {
485 t.Fatal(err)
486 }
487 // Wait for the runner to actually start.
488 deadline := time.Now().Add(time.Second)
489 for a.Status() != StatusRunning && time.Now().Before(deadline) {
490 time.Sleep(2 * time.Millisecond)
491 }
492 if _, err := f.Resume(context.Background(), a.ID); err == nil {
493 t.Fatal("resume on running agent did not error")
494 }
495}
496
497// TestResumeAfterReload exercises the full lifecycle the user cares
498// about: spawn in process A, exit A, start process B, Reload, Resume.
499// The resumed agent must keep its id and start producing new output
500// against the same on-disk state.
501//
502// Note on transcript persistence: in production the runner is
503// execRunner, which writes every event to events.jsonl, and Reload's
504// replayEventsIntoAgent rebuilds the transcript from there. The fake
505// RunnerFunc in this test calls sink.Transcript directly (which only
506// touches in-memory state), so the "first run" line is *expected* to
507// be lost across the restart — the in-memory Agent went away. We
508// have TestReloadReplaysTranscriptFromEventLog covering the
509// log-replay path with a curated events.jsonl.
510func TestResumeAfterReload(t *testing.T) {
511 root := t.TempDir()
512 // Process A
513 a := New(Config{
514 Root: root, RepoRoot: root,
515 NewRunner: func(ag *Agent) Runner {
516 return RunnerFunc(func(ctx context.Context, sink Sink) error {
517 sink.Transcript("first run for " + ag.ID)
518 <-ctx.Done()
519 return ctx.Err()
520 })
521 },
522 })
523 ag, err := a.Spawn(context.Background(), "x")
524 if err != nil {
525 t.Fatal(err)
526 }
527 // Let the runner record its transcript line before we tear it down.
528 deadline := time.Now().Add(time.Second)
529 for len(ag.Transcript()) == 0 && time.Now().Before(deadline) {
530 time.Sleep(2 * time.Millisecond)
531 }
532 id := ag.ID
533 a.StopAll()
534 ag.Wait()
535
536 // Process B: a new Swarm against the same root.
537 resumed := make(chan struct{}, 1)
538 b := New(Config{
539 Root: root, RepoRoot: root,
540 NewRunner: func(ag *Agent) Runner {
541 return RunnerFunc(func(ctx context.Context, sink Sink) error {
542 sink.Transcript("second run for " + ag.ID)
543 select {
544 case resumed <- struct{}{}:
545 default:
546 }
547 <-ctx.Done()
548 return ctx.Err()
549 })
550 },
551 })
552 if loaded, errs := b.Reload(); loaded != 1 || len(errs) > 0 {
553 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
554 }
555 a2, err := b.Resume(context.Background(), id)
556 if err != nil {
557 t.Fatalf("resume: %v", err)
558 }
559 defer b.StopAll()
560
561 select {
562 case <-resumed:
563 case <-time.After(time.Second):
564 t.Fatal("resume runner did not start")
565 }
566 if a2.ID != id {
567 t.Errorf("resume id = %s; want %s", a2.ID, id)
568 }
569
570 // Wait for the new transcript line to land.
571 deadline = time.Now().Add(time.Second)
572 for time.Now().Before(deadline) {
573 for _, l := range a2.Transcript() {
574 if l == "second run for "+id {
575 return
576 }
577 }
578 time.Sleep(5 * time.Millisecond)
579 }
580 t.Fatalf("resumed transcript missing fresh line: %v", a2.Transcript())
581}
582
583// TestSpawnReqPersistsModel confirms that the per-agent model
584// override is captured at Spawn time, surfaced via Snapshot, and
585// written to meta.json so a later Reload + Resume reuses it.
586func TestSpawnReqPersistsModel(t *testing.T) {
587 root := t.TempDir()
588 f := New(Config{
589 Root: root, RepoRoot: root,
590 NewRunner: func(a *Agent) Runner {
591 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
592 },
593 })
594 a, err := f.SpawnReq(context.Background(), SpawnRequest{
595 Task: "x", Model: "claude-sonnet-4-5", Provider: "anthropic",
596 })
597 if err != nil {
598 t.Fatalf("spawn: %v", err)
599 }
600 if a.Model != "claude-sonnet-4-5" || a.Provider != "anthropic" {
601 t.Fatalf("agent fields = (%q,%q); want (claude-sonnet-4-5, anthropic)", a.Model, a.Provider)
602 }
603 snap := a.Snapshot()
604 if snap.Model != "claude-sonnet-4-5" || snap.Provider != "anthropic" {
605 t.Fatalf("snapshot = (%q,%q); model fields not surfaced", snap.Model, snap.Provider)
606 }
607
608 // Stop so we can read meta.json without racing the run loop.
609 if err := f.Stop(a.ID); err != nil {
610 t.Fatal(err)
611 }
612 a.Wait()
613
614 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
615 if err != nil {
616 t.Fatalf("read meta.json: %v", err)
617 }
618 var got agentMeta
619 if err := json.Unmarshal(metaBytes, &got); err != nil {
620 t.Fatal(err)
621 }
622 if got.Model != "claude-sonnet-4-5" || got.Provider != "anthropic" {
623 t.Errorf("meta = (%q,%q); want model + provider persisted", got.Model, got.Provider)
624 }
625
626 // Reload in a fresh Swarm and confirm the detached agent still
627 // carries the model/provider so Resume can route the child
628 // subprocess back to the same model.
629 g := New(Config{Root: root, RepoRoot: root})
630 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
631 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
632 }
633 re := g.Get(a.ID)
634 if re == nil {
635 t.Fatal("reloaded agent missing")
636 }
637 if re.Model != "claude-sonnet-4-5" || re.Provider != "anthropic" {
638 t.Errorf("reloaded fields = (%q,%q); want preserved", re.Model, re.Provider)
639 }
640}
641
642// TestSwarmAgentArgsIncludesModelFlags pins the argv contract that
643// connects the supervisor's per-agent model override to the child's
644// --model / --provider flag set. Adding a new path to argv without
645// updating this assertion is the failure mode this catches.
646func TestSwarmAgentArgsIncludesModelFlags(t *testing.T) {
647 args := swarmAgentArgs(swarmAgentArgsOpts{
648 Exe: "/zot", Dir: "/wt", SessionPath: "/s.json", InboxPath: "/in.sock",
649 Task: "do x", Model: "gpt-5", Provider: "openai",
650 })
651 want := []string{"--model", "gpt-5", "--provider", "openai"}
652 for i := 0; i+1 < len(want); i += 2 {
653 flag, value := want[i], want[i+1]
654 found := false
655 for j := 0; j+1 < len(args); j++ {
656 if args[j] == flag && args[j+1] == value {
657 found = true
658 break
659 }
660 }
661 if !found {
662 t.Errorf("argv missing %s %s: %v", flag, value, args)
663 }
664 }
665 // And the task still ends up positional last.
666 if args[len(args)-1] != "do x" {
667 t.Errorf("task should be last positional; argv=%v", args)
668 }
669}
670
671// TestStopOnDetachedAgentIsNoopAndDoesNotPanic regression-tests the
672// segfault that hit when the user pressed 'k' on a reloaded
673// (detached) agent: Swarm.Stop unconditionally called a.cancel(),
674// but buildDetachedAgent never assigns a cancel func because there's
675// no in-process runner to cancel. The fix is to short-circuit Stop
676// for StatusDetached and — belt-and-braces — nil-check a.cancel.
677func TestStopOnDetachedAgentIsNoopAndDoesNotPanic(t *testing.T) {
678 root := t.TempDir()
679 // Build a detached agent the same way Reload does: drop a
680 // meta.json on disk, then ask a fresh Swarm to pick it up.
681 id := "alpha-1"
682 stateDir := filepath.Join(root, "agents", id)
683 if err := os.MkdirAll(stateDir, 0o755); err != nil {
684 t.Fatal(err)
685 }
686 m := agentMeta{
687 ID: id, Task: "t",
688 Dir: root,
689 Started: time.Now().Add(-time.Hour),
690 InboxPath: filepath.Join(stateDir, "in.sock"),
691 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
692 SessionPath: filepath.Join(stateDir, "session.json"),
693 }
694 mb, _ := json.MarshalIndent(m, "", " ")
695 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
696 t.Fatal(err)
697 }
698
699 f := New(Config{Root: root, RepoRoot: root})
700 if loaded, errs := f.Reload(); loaded != 1 || len(errs) > 0 {
701 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
702 }
703 a := f.Get(id)
704 if a == nil {
705 t.Fatal("detached agent missing from swarm")
706 }
707 if a.Status() != StatusDetached {
708 t.Fatalf("setup: agent status = %q; want detached", a.Status())
709 }
710
711 // The real assertion: Stop returns cleanly without panicking.
712 defer func() {
713 if r := recover(); r != nil {
714 t.Fatalf("Stop on detached agent panicked: %v", r)
715 }
716 }()
717 if err := f.Stop(id); err != nil {
718 t.Fatalf("Stop on detached agent: %v", err)
719 }
720 // Status must remain Detached (Stop is a no-op).
721 if got := a.Status(); got != StatusDetached {
722 t.Errorf("after Stop status = %q; want detached (no-op)", got)
723 }
724
725 // StopAll, which the cli defers on exit, must also be safe.
726 f.StopAll()
727}
728
729// TestRemoveAlsoCleansStateDir ensures Remove deletes the on-disk
730// state directory in addition to the worktree, so a removed agent
731// doesn't reappear on the next Reload.
732func TestRemoveAlsoCleansStateDir(t *testing.T) {
733 root := t.TempDir()
734 f := New(Config{
735 Root: root, RepoRoot: root,
736 NewRunner: func(a *Agent) Runner {
737 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
738 },
739 })
740 a, err := f.Spawn(context.Background(), "x")
741 if err != nil {
742 t.Fatal(err)
743 }
744 stateDir := filepath.Join(root, "agents", a.ID)
745 if _, err := os.Stat(filepath.Join(stateDir, "meta.json")); err != nil {
746 t.Fatalf("meta.json missing pre-remove: %v", err)
747 }
748 if err := f.Stop(a.ID); err != nil {
749 t.Fatal(err)
750 }
751 a.Wait()
752 if err := f.Remove(a.ID); err != nil {
753 t.Fatal(err)
754 }
755 if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
756 t.Fatalf("state dir should be gone after remove; got %v", err)
757 }
758
759 // A fresh Swarm + Reload should find nothing.
760 g := New(Config{Root: root, RepoRoot: root})
761 if loaded, _ := g.Reload(); loaded != 0 {
762 t.Fatalf("reload after remove loaded=%d; want 0", loaded)
763 }
764}
765
766// TestActiveSessionScopesSnapshotAll proves the session filter the
767// user asked for: a single Swarm with agents spawned under two
768// different active sessions only surfaces the agents matching the
769// currently-active session via SnapshotAll. Switching the active
770// session re-narrows the view without touching agent state.
771func TestActiveSessionScopesSnapshotAll(t *testing.T) {
772 root := t.TempDir()
773 f := New(Config{
774 Root: root, RepoRoot: root,
775 NewRunner: func(a *Agent) Runner {
776 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
777 },
778 })
779
780 // Spawn one agent under session A and one under session B.
781 f.SetActiveSession("sess-A")
782 aA, err := f.Spawn(context.Background(), "task A")
783 if err != nil {
784 t.Fatal(err)
785 }
786 f.SetActiveSession("sess-B")
787 aB, err := f.Spawn(context.Background(), "task B")
788 if err != nil {
789 t.Fatal(err)
790 }
791
792 // Both agents must carry the session id they were spawned under,
793 // regardless of any later SetActiveSession.
794 if aA.SessionID != "sess-A" {
795 t.Errorf("aA.SessionID = %q; want sess-A", aA.SessionID)
796 }
797 if aB.SessionID != "sess-B" {
798 t.Errorf("aB.SessionID = %q; want sess-B", aB.SessionID)
799 }
800
801 // Active session B → only aB visible.
802 only := snapshotIDs(f.SnapshotAll())
803 if len(only) != 1 || only[0] != aB.ID {
804 t.Errorf("scoped to sess-B, snapshot ids = %v; want [%s]", only, aB.ID)
805 }
806
807 // Switch back to A → only aA visible.
808 f.SetActiveSession("sess-A")
809 only = snapshotIDs(f.SnapshotAll())
810 if len(only) != 1 || only[0] != aA.ID {
811 t.Errorf("scoped to sess-A, snapshot ids = %v; want [%s]", only, aA.ID)
812 }
813
814 // Clear the scope → both visible.
815 f.SetActiveSession("")
816 all := snapshotIDs(f.SnapshotAll())
817 if len(all) != 2 {
818 t.Errorf("unscoped snapshot ids = %v; want both agents", all)
819 }
820
821 // Cleanup.
822 _ = f.Stop(aA.ID)
823 _ = f.Stop(aB.ID)
824 aA.Wait()
825 aB.Wait()
826}
827
828// TestSessionIDPersistsAcrossReload confirms the SessionID field
829// survives a Swarm restart: an agent spawned under session A in one
830// Swarm instance is still SessionID="sess-A" after a fresh New +
831// Reload reads it back from meta.json. Without persistence, the
832// scope filter would forget which session owned each agent after
833// a zot restart and the dashboard would leak everything again.
834func TestSessionIDPersistsAcrossReload(t *testing.T) {
835 root := t.TempDir()
836 mkSwarm := func() *Swarm {
837 return New(Config{
838 Root: root, RepoRoot: root,
839 NewRunner: func(a *Agent) Runner {
840 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
841 },
842 })
843 }
844
845 f := mkSwarm()
846 f.SetActiveSession("sess-keep")
847 a, err := f.Spawn(context.Background(), "persist me")
848 if err != nil {
849 t.Fatal(err)
850 }
851 _ = f.Stop(a.ID)
852 a.Wait()
853
854 // Fresh Swarm, same root, reload from disk.
855 g := mkSwarm()
856 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
857 t.Fatalf("reload loaded=%d errs=%v; want 1 / no errs", loaded, errs)
858 }
859 g.SetActiveSession("sess-keep")
860 got := snapshotIDs(g.SnapshotAll())
861 if len(got) != 1 || got[0] != a.ID {
862 t.Errorf("after reload + scope to sess-keep, ids = %v; want [%s]", got, a.ID)
863 }
864
865 // Scope to a different session: agent must be hidden.
866 g.SetActiveSession("sess-other")
867 if got := snapshotIDs(g.SnapshotAll()); len(got) != 0 {
868 t.Errorf("scoped to other session, ids = %v; want []", got)
869 }
870}
871
872// TestEmptySessionIDIsVisibleFromAnyScope pins the backward-compat
873// rule: agents whose meta.json predates the SessionID field (or who
874// were spawned without an active session, e.g. via a test rig or
875// scripted caller) carry SessionID == "" and remain visible from
876// every scope. Otherwise the schema bump would orphan every
877// pre-existing agent the moment a user upgraded zot.
878func TestEmptySessionIDIsVisibleFromAnyScope(t *testing.T) {
879 root := t.TempDir()
880 f := New(Config{
881 Root: root, RepoRoot: root,
882 NewRunner: func(a *Agent) Runner {
883 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
884 },
885 })
886 // No SetActiveSession call → agent spawned with empty SessionID.
887 a, err := f.Spawn(context.Background(), "legacy")
888 if err != nil {
889 t.Fatal(err)
890 }
891 if a.SessionID != "" {
892 t.Fatalf("unscoped spawn produced SessionID %q; want empty", a.SessionID)
893 }
894
895 for _, scope := range []string{"", "any-session", "some-other"} {
896 f.SetActiveSession(scope)
897 got := snapshotIDs(f.SnapshotAll())
898 if len(got) != 1 || got[0] != a.ID {
899 t.Errorf("scope=%q: ids = %v; want legacy agent visible", scope, got)
900 }
901 }
902
903 _ = f.Stop(a.ID)
904 a.Wait()
905}
906
907func snapshotIDs(ss []AgentSnapshot) []string {
908 out := make([]string, len(ss))
909 for i, s := range ss {
910 out[i] = s.ID
911 }
912 return out
913}