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
224// TestReloadSkipsBareDirsAndCorruptMeta ensures one bad meta.json
225// doesn't blow up the whole reload. Directories with no meta.json
226// at all are silently ignored (Spawn that failed mid-way leaves
227// these); corrupt meta files are reported as errors but don't stop
228// the rest of the load.
229func TestReloadSkipsBareDirsAndCorruptMeta(t *testing.T) {
230 root := t.TempDir()
231 agentsDir := filepath.Join(root, "agents")
232 // Bare directory, no meta.json.
233 if err := os.MkdirAll(filepath.Join(agentsDir, "leftover"), 0o755); err != nil {
234 t.Fatal(err)
235 }
236 // Directory with garbage meta.json.
237 if err := os.MkdirAll(filepath.Join(agentsDir, "corrupt"), 0o755); err != nil {
238 t.Fatal(err)
239 }
240 if err := os.WriteFile(filepath.Join(agentsDir, "corrupt", "meta.json"), []byte("not json"), 0o644); err != nil {
241 t.Fatal(err)
242 }
243 // One good agent.
244 good := "good-1"
245 stateDir := filepath.Join(agentsDir, good)
246 _ = os.MkdirAll(stateDir, 0o755)
247 m := agentMeta{ID: good, Task: "x", Dir: "/tmp/x", Started: time.Now()}
248 mb, _ := json.MarshalIndent(m, "", " ")
249 _ = os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644)
250
251 f := New(Config{Root: root, RepoRoot: root})
252 loaded, errs := f.Reload()
253 if loaded != 1 {
254 t.Errorf("loaded = %d; want 1", loaded)
255 }
256 if len(errs) != 1 {
257 t.Errorf("errs len = %d; want 1 (the corrupt entry): %v", len(errs), errs)
258 }
259}
260
261// TestResumeRestartsRunnerOnSameSession is the headline test. We:
262//
263// 1. Spawn an agent with a counting runner (records each Run call).
264// 2. Stop it.
265// 3. Resume it.
266// 4. Assert Run was called twice — once for spawn, once for resume —
267// against the same SessionPath / InboxPath / Dir.
268func TestResumeRestartsRunnerOnSameSession(t *testing.T) {
269 root := t.TempDir()
270 var (
271 mu sync.Mutex
272 runs []*Agent
273 runsDone int32
274 )
275 f := New(Config{
276 Root: root, RepoRoot: root,
277 NewRunner: func(a *Agent) Runner {
278 return RunnerFunc(func(ctx context.Context, sink Sink) error {
279 mu.Lock()
280 runs = append(runs, a)
281 mu.Unlock()
282 sink.Activity("ran")
283 atomic.AddInt32(&runsDone, 1)
284 <-ctx.Done()
285 return ctx.Err()
286 })
287 },
288 })
289 defer f.StopAll()
290
291 a, err := f.Spawn(context.Background(), "do thing")
292 if err != nil {
293 t.Fatalf("spawn: %v", err)
294 }
295 originalSession := a.SessionPath
296 originalDir := a.Dir
297 originalInbox := a.InboxPath
298
299 // Wait for the runner to record its presence, then stop the agent.
300 deadline := time.Now().Add(time.Second)
301 for atomic.LoadInt32(&runsDone) < 1 && time.Now().Before(deadline) {
302 time.Sleep(2 * time.Millisecond)
303 }
304 if err := f.Stop(a.ID); err != nil {
305 t.Fatalf("stop: %v", err)
306 }
307 a.Wait()
308 if a.Status() != StatusKilled {
309 t.Fatalf("post-stop status = %q; want killed", a.Status())
310 }
311
312 // Now resume.
313 a2, err := f.Resume(context.Background(), a.ID)
314 if err != nil {
315 t.Fatalf("resume: %v", err)
316 }
317 if a2.ID != a.ID {
318 t.Errorf("resume returned different id: %s vs %s", a2.ID, a.ID)
319 }
320 if a2.SessionPath != originalSession {
321 t.Errorf("resume changed session path: %s vs %s", a2.SessionPath, originalSession)
322 }
323 if a2.Dir != originalDir {
324 t.Errorf("resume changed dir: %s vs %s", a2.Dir, originalDir)
325 }
326 if a2.InboxPath != originalInbox {
327 t.Errorf("resume changed inbox path: %s vs %s", a2.InboxPath, originalInbox)
328 }
329 // Two runner invocations: spawn + resume.
330 deadline = time.Now().Add(time.Second)
331 for atomic.LoadInt32(&runsDone) < 2 && time.Now().Before(deadline) {
332 time.Sleep(2 * time.Millisecond)
333 }
334 mu.Lock()
335 defer mu.Unlock()
336 if len(runs) != 2 {
337 t.Fatalf("runner invocations = %d; want 2", len(runs))
338 }
339 if runs[1].ID != a.ID {
340 t.Errorf("second run had wrong id: %s", runs[1].ID)
341 }
342 if got := len(f.SnapshotAll()); got != 1 {
343 t.Errorf("snapshot len = %d; want 1 (in-place replace)", got)
344 }
345}
346
347func TestResumeRecalculatesLegacyStateInboxPath(t *testing.T) {
348 if runtime.GOOS == "windows" {
349 t.Skip("swarm inbox runtime relocation applies to Unix sockets")
350 }
351 root := t.TempDir()
352 f := New(Config{
353 Root: root, RepoRoot: root,
354 NewRunner: func(*Agent) Runner {
355 return RunnerFunc(func(context.Context, Sink) error { return nil })
356 },
357 })
358 a, err := f.Spawn(context.Background(), "legacy inbox")
359 if err != nil {
360 t.Fatal(err)
361 }
362 a.Wait()
363
364 legacyPath := filepath.Join(root, "agents", a.ID, "in.sock")
365 a.InboxPath = legacyPath
366 a2, err := f.Resume(context.Background(), a.ID)
367 if err != nil {
368 t.Fatalf("resume: %v", err)
369 }
370 a2.Wait()
371 if a2.InboxPath == legacyPath {
372 t.Fatalf("resume kept legacy state-directory inbox path %q", legacyPath)
373 }
374 if strings.HasPrefix(a2.InboxPath, root+string(filepath.Separator)) {
375 t.Fatalf("resumed inbox %q remains below state root %q", a2.InboxPath, root)
376 }
377}
378
379// TestResumeSetsResumingFlag pins the contract execRunner relies on:
380// after Resume, the new Agent has Resuming == true; after a fresh
381// Spawn it's false. The runner branches on this to decide whether
382// to pass the original Task as a positional argv to the child, so
383// getting it wrong silently re-runs the initial task on every
384// resume and surfaces "agent busy; send 'cancel' first" between
385// assistant messages.
386func TestResumeSetsResumingFlag(t *testing.T) {
387 root := t.TempDir()
388 f := New(Config{
389 Root: root, RepoRoot: root,
390 NewRunner: func(a *Agent) Runner {
391 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
392 },
393 })
394 defer f.StopAll()
395
396 a, err := f.Spawn(context.Background(), "x")
397 if err != nil {
398 t.Fatal(err)
399 }
400 if a.Resuming {
401 t.Fatal("fresh Spawn produced Resuming==true; want false")
402 }
403 if err := f.Stop(a.ID); err != nil {
404 t.Fatal(err)
405 }
406 a.Wait()
407
408 a2, err := f.Resume(context.Background(), a.ID)
409 if err != nil {
410 t.Fatalf("resume: %v", err)
411 }
412 if !a2.Resuming {
413 t.Error("Resume produced Resuming==false; want true so the runner skips the duplicate initial-task argv")
414 }
415}
416
417// TestResumeRejectsRunningAgent prevents the user from double-running
418// an agent: two runners on the same session.json would race.
419func TestResumeRejectsRunningAgent(t *testing.T) {
420 root := t.TempDir()
421 f := New(Config{
422 Root: root, RepoRoot: root,
423 NewRunner: func(a *Agent) Runner {
424 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
425 },
426 })
427 defer f.StopAll()
428 a, err := f.Spawn(context.Background(), "x")
429 if err != nil {
430 t.Fatal(err)
431 }
432 // Wait for the runner to actually start.
433 deadline := time.Now().Add(time.Second)
434 for a.Status() != StatusRunning && time.Now().Before(deadline) {
435 time.Sleep(2 * time.Millisecond)
436 }
437 if _, err := f.Resume(context.Background(), a.ID); err == nil {
438 t.Fatal("resume on running agent did not error")
439 }
440}
441
442// TestResumeAfterReload exercises the full lifecycle the user cares
443// about: spawn in process A, exit A, start process B, Reload, Resume.
444// The resumed agent must keep its id and start producing new output
445// against the same on-disk state.
446//
447// Note on transcript persistence: in production the runner is
448// execRunner, which writes every event to events.jsonl, and Reload's
449// replayEventsIntoAgent rebuilds the transcript from there. The fake
450// RunnerFunc in this test calls sink.Transcript directly (which only
451// touches in-memory state), so the "first run" line is *expected* to
452// be lost across the restart — the in-memory Agent went away. We
453// have TestReloadReplaysTranscriptFromEventLog covering the
454// log-replay path with a curated events.jsonl.
455func TestResumeAfterReload(t *testing.T) {
456 root := t.TempDir()
457 // Process A
458 a := New(Config{
459 Root: root, RepoRoot: root,
460 NewRunner: func(ag *Agent) Runner {
461 return RunnerFunc(func(ctx context.Context, sink Sink) error {
462 sink.Transcript("first run for " + ag.ID)
463 <-ctx.Done()
464 return ctx.Err()
465 })
466 },
467 })
468 ag, err := a.Spawn(context.Background(), "x")
469 if err != nil {
470 t.Fatal(err)
471 }
472 // Let the runner record its transcript line before we tear it down.
473 deadline := time.Now().Add(time.Second)
474 for len(ag.Transcript()) == 0 && time.Now().Before(deadline) {
475 time.Sleep(2 * time.Millisecond)
476 }
477 id := ag.ID
478 a.StopAll()
479 ag.Wait()
480
481 // Process B: a new Swarm against the same root.
482 resumed := make(chan struct{}, 1)
483 b := New(Config{
484 Root: root, RepoRoot: root,
485 NewRunner: func(ag *Agent) Runner {
486 return RunnerFunc(func(ctx context.Context, sink Sink) error {
487 sink.Transcript("second run for " + ag.ID)
488 select {
489 case resumed <- struct{}{}:
490 default:
491 }
492 <-ctx.Done()
493 return ctx.Err()
494 })
495 },
496 })
497 if loaded, errs := b.Reload(); loaded != 1 || len(errs) > 0 {
498 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
499 }
500 a2, err := b.Resume(context.Background(), id)
501 if err != nil {
502 t.Fatalf("resume: %v", err)
503 }
504 defer b.StopAll()
505
506 select {
507 case <-resumed:
508 case <-time.After(time.Second):
509 t.Fatal("resume runner did not start")
510 }
511 if a2.ID != id {
512 t.Errorf("resume id = %s; want %s", a2.ID, id)
513 }
514
515 // Wait for the new transcript line to land.
516 deadline = time.Now().Add(time.Second)
517 for time.Now().Before(deadline) {
518 for _, l := range a2.Transcript() {
519 if l == "second run for "+id {
520 return
521 }
522 }
523 time.Sleep(5 * time.Millisecond)
524 }
525 t.Fatalf("resumed transcript missing fresh line: %v", a2.Transcript())
526}
527
528// TestSpawnReqPersistsModel confirms that the per-agent model
529// override is captured at Spawn time, surfaced via Snapshot, and
530// written to meta.json so a later Reload + Resume reuses it.
531func TestSpawnReqPersistsModel(t *testing.T) {
532 root := t.TempDir()
533 f := New(Config{
534 Root: root, RepoRoot: root,
535 NewRunner: func(a *Agent) Runner {
536 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
537 },
538 })
539 a, err := f.SpawnReq(context.Background(), SpawnRequest{
540 Task: "x", Model: "claude-sonnet-4-5", Provider: "anthropic",
541 })
542 if err != nil {
543 t.Fatalf("spawn: %v", err)
544 }
545 if a.Model != "claude-sonnet-4-5" || a.Provider != "anthropic" {
546 t.Fatalf("agent fields = (%q,%q); want (claude-sonnet-4-5, anthropic)", a.Model, a.Provider)
547 }
548 snap := a.Snapshot()
549 if snap.Model != "claude-sonnet-4-5" || snap.Provider != "anthropic" {
550 t.Fatalf("snapshot = (%q,%q); model fields not surfaced", snap.Model, snap.Provider)
551 }
552
553 // Stop so we can read meta.json without racing the run loop.
554 if err := f.Stop(a.ID); err != nil {
555 t.Fatal(err)
556 }
557 a.Wait()
558
559 metaBytes, err := os.ReadFile(filepath.Join(root, "agents", a.ID, "meta.json"))
560 if err != nil {
561 t.Fatalf("read meta.json: %v", err)
562 }
563 var got agentMeta
564 if err := json.Unmarshal(metaBytes, &got); err != nil {
565 t.Fatal(err)
566 }
567 if got.Model != "claude-sonnet-4-5" || got.Provider != "anthropic" {
568 t.Errorf("meta = (%q,%q); want model + provider persisted", got.Model, got.Provider)
569 }
570
571 // Reload in a fresh Swarm and confirm the detached agent still
572 // carries the model/provider so Resume can route the child
573 // subprocess back to the same model.
574 g := New(Config{Root: root, RepoRoot: root})
575 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
576 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
577 }
578 re := g.Get(a.ID)
579 if re == nil {
580 t.Fatal("reloaded agent missing")
581 }
582 if re.Model != "claude-sonnet-4-5" || re.Provider != "anthropic" {
583 t.Errorf("reloaded fields = (%q,%q); want preserved", re.Model, re.Provider)
584 }
585}
586
587// TestSwarmAgentArgsIncludesModelFlags pins the argv contract that
588// connects the supervisor's per-agent model override to the child's
589// --model / --provider flag set. Adding a new path to argv without
590// updating this assertion is the failure mode this catches.
591func TestSwarmAgentArgsIncludesModelFlags(t *testing.T) {
592 args := swarmAgentArgs(swarmAgentArgsOpts{
593 Exe: "/zot", Dir: "/wt", SessionPath: "/s.json", InboxPath: "/in.sock",
594 Task: "do x", Model: "gpt-5", Provider: "openai",
595 })
596 want := []string{"--model", "gpt-5", "--provider", "openai"}
597 for i := 0; i+1 < len(want); i += 2 {
598 flag, value := want[i], want[i+1]
599 found := false
600 for j := 0; j+1 < len(args); j++ {
601 if args[j] == flag && args[j+1] == value {
602 found = true
603 break
604 }
605 }
606 if !found {
607 t.Errorf("argv missing %s %s: %v", flag, value, args)
608 }
609 }
610 // And the task still ends up positional last.
611 if args[len(args)-1] != "do x" {
612 t.Errorf("task should be last positional; argv=%v", args)
613 }
614}
615
616// TestStopOnDetachedAgentIsNoopAndDoesNotPanic regression-tests the
617// segfault that hit when the user pressed 'k' on a reloaded
618// (detached) agent: Swarm.Stop unconditionally called a.cancel(),
619// but buildDetachedAgent never assigns a cancel func because there's
620// no in-process runner to cancel. The fix is to short-circuit Stop
621// for StatusDetached and — belt-and-braces — nil-check a.cancel.
622func TestStopOnDetachedAgentIsNoopAndDoesNotPanic(t *testing.T) {
623 root := t.TempDir()
624 // Build a detached agent the same way Reload does: drop a
625 // meta.json on disk, then ask a fresh Swarm to pick it up.
626 id := "alpha-1"
627 stateDir := filepath.Join(root, "agents", id)
628 if err := os.MkdirAll(stateDir, 0o755); err != nil {
629 t.Fatal(err)
630 }
631 m := agentMeta{
632 ID: id, Task: "t",
633 Dir: root,
634 Started: time.Now().Add(-time.Hour),
635 InboxPath: filepath.Join(stateDir, "in.sock"),
636 EventLogPath: filepath.Join(stateDir, "events.jsonl"),
637 SessionPath: filepath.Join(stateDir, "session.json"),
638 }
639 mb, _ := json.MarshalIndent(m, "", " ")
640 if err := os.WriteFile(filepath.Join(stateDir, "meta.json"), mb, 0o644); err != nil {
641 t.Fatal(err)
642 }
643
644 f := New(Config{Root: root, RepoRoot: root})
645 if loaded, errs := f.Reload(); loaded != 1 || len(errs) > 0 {
646 t.Fatalf("reload loaded=%d errs=%v", loaded, errs)
647 }
648 a := f.Get(id)
649 if a == nil {
650 t.Fatal("detached agent missing from swarm")
651 }
652 if a.Status() != StatusDetached {
653 t.Fatalf("setup: agent status = %q; want detached", a.Status())
654 }
655
656 // The real assertion: Stop returns cleanly without panicking.
657 defer func() {
658 if r := recover(); r != nil {
659 t.Fatalf("Stop on detached agent panicked: %v", r)
660 }
661 }()
662 if err := f.Stop(id); err != nil {
663 t.Fatalf("Stop on detached agent: %v", err)
664 }
665 // Status must remain Detached (Stop is a no-op).
666 if got := a.Status(); got != StatusDetached {
667 t.Errorf("after Stop status = %q; want detached (no-op)", got)
668 }
669
670 // StopAll, which the cli defers on exit, must also be safe.
671 f.StopAll()
672}
673
674// TestRemoveAlsoCleansStateDir ensures Remove deletes the on-disk
675// state directory in addition to the worktree, so a removed agent
676// doesn't reappear on the next Reload.
677func TestRemoveAlsoCleansStateDir(t *testing.T) {
678 root := t.TempDir()
679 f := New(Config{
680 Root: root, RepoRoot: root,
681 NewRunner: func(a *Agent) Runner {
682 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
683 },
684 })
685 a, err := f.Spawn(context.Background(), "x")
686 if err != nil {
687 t.Fatal(err)
688 }
689 stateDir := filepath.Join(root, "agents", a.ID)
690 if _, err := os.Stat(filepath.Join(stateDir, "meta.json")); err != nil {
691 t.Fatalf("meta.json missing pre-remove: %v", err)
692 }
693 if err := f.Stop(a.ID); err != nil {
694 t.Fatal(err)
695 }
696 a.Wait()
697 if err := f.Remove(a.ID); err != nil {
698 t.Fatal(err)
699 }
700 if _, err := os.Stat(stateDir); !errors.Is(err, os.ErrNotExist) {
701 t.Fatalf("state dir should be gone after remove; got %v", err)
702 }
703
704 // A fresh Swarm + Reload should find nothing.
705 g := New(Config{Root: root, RepoRoot: root})
706 if loaded, _ := g.Reload(); loaded != 0 {
707 t.Fatalf("reload after remove loaded=%d; want 0", loaded)
708 }
709}
710
711// TestActiveSessionScopesSnapshotAll proves the session filter the
712// user asked for: a single Swarm with agents spawned under two
713// different active sessions only surfaces the agents matching the
714// currently-active session via SnapshotAll. Switching the active
715// session re-narrows the view without touching agent state.
716func TestActiveSessionScopesSnapshotAll(t *testing.T) {
717 root := t.TempDir()
718 f := New(Config{
719 Root: root, RepoRoot: root,
720 NewRunner: func(a *Agent) Runner {
721 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
722 },
723 })
724
725 // Spawn one agent under session A and one under session B.
726 f.SetActiveSession("sess-A")
727 aA, err := f.Spawn(context.Background(), "task A")
728 if err != nil {
729 t.Fatal(err)
730 }
731 f.SetActiveSession("sess-B")
732 aB, err := f.Spawn(context.Background(), "task B")
733 if err != nil {
734 t.Fatal(err)
735 }
736
737 // Both agents must carry the session id they were spawned under,
738 // regardless of any later SetActiveSession.
739 if aA.SessionID != "sess-A" {
740 t.Errorf("aA.SessionID = %q; want sess-A", aA.SessionID)
741 }
742 if aB.SessionID != "sess-B" {
743 t.Errorf("aB.SessionID = %q; want sess-B", aB.SessionID)
744 }
745
746 // Active session B → only aB visible.
747 only := snapshotIDs(f.SnapshotAll())
748 if len(only) != 1 || only[0] != aB.ID {
749 t.Errorf("scoped to sess-B, snapshot ids = %v; want [%s]", only, aB.ID)
750 }
751
752 // Switch back to A → only aA visible.
753 f.SetActiveSession("sess-A")
754 only = snapshotIDs(f.SnapshotAll())
755 if len(only) != 1 || only[0] != aA.ID {
756 t.Errorf("scoped to sess-A, snapshot ids = %v; want [%s]", only, aA.ID)
757 }
758
759 // Clear the scope → both visible.
760 f.SetActiveSession("")
761 all := snapshotIDs(f.SnapshotAll())
762 if len(all) != 2 {
763 t.Errorf("unscoped snapshot ids = %v; want both agents", all)
764 }
765
766 // Cleanup.
767 _ = f.Stop(aA.ID)
768 _ = f.Stop(aB.ID)
769 aA.Wait()
770 aB.Wait()
771}
772
773// TestSessionIDPersistsAcrossReload confirms the SessionID field
774// survives a Swarm restart: an agent spawned under session A in one
775// Swarm instance is still SessionID="sess-A" after a fresh New +
776// Reload reads it back from meta.json. Without persistence, the
777// scope filter would forget which session owned each agent after
778// a zot restart and the dashboard would leak everything again.
779func TestSessionIDPersistsAcrossReload(t *testing.T) {
780 root := t.TempDir()
781 mkSwarm := func() *Swarm {
782 return New(Config{
783 Root: root, RepoRoot: root,
784 NewRunner: func(a *Agent) Runner {
785 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
786 },
787 })
788 }
789
790 f := mkSwarm()
791 f.SetActiveSession("sess-keep")
792 a, err := f.Spawn(context.Background(), "persist me")
793 if err != nil {
794 t.Fatal(err)
795 }
796 _ = f.Stop(a.ID)
797 a.Wait()
798
799 // Fresh Swarm, same root, reload from disk.
800 g := mkSwarm()
801 if loaded, errs := g.Reload(); loaded != 1 || len(errs) > 0 {
802 t.Fatalf("reload loaded=%d errs=%v; want 1 / no errs", loaded, errs)
803 }
804 g.SetActiveSession("sess-keep")
805 got := snapshotIDs(g.SnapshotAll())
806 if len(got) != 1 || got[0] != a.ID {
807 t.Errorf("after reload + scope to sess-keep, ids = %v; want [%s]", got, a.ID)
808 }
809
810 // Scope to a different session: agent must be hidden.
811 g.SetActiveSession("sess-other")
812 if got := snapshotIDs(g.SnapshotAll()); len(got) != 0 {
813 t.Errorf("scoped to other session, ids = %v; want []", got)
814 }
815}
816
817// TestEmptySessionIDIsVisibleFromAnyScope pins the backward-compat
818// rule: agents whose meta.json predates the SessionID field (or who
819// were spawned without an active session, e.g. via a test rig or
820// scripted caller) carry SessionID == "" and remain visible from
821// every scope. Otherwise the schema bump would orphan every
822// pre-existing agent the moment a user upgraded zot.
823func TestEmptySessionIDIsVisibleFromAnyScope(t *testing.T) {
824 root := t.TempDir()
825 f := New(Config{
826 Root: root, RepoRoot: root,
827 NewRunner: func(a *Agent) Runner {
828 return RunnerFunc(func(ctx context.Context, _ Sink) error { <-ctx.Done(); return ctx.Err() })
829 },
830 })
831 // No SetActiveSession call → agent spawned with empty SessionID.
832 a, err := f.Spawn(context.Background(), "legacy")
833 if err != nil {
834 t.Fatal(err)
835 }
836 if a.SessionID != "" {
837 t.Fatalf("unscoped spawn produced SessionID %q; want empty", a.SessionID)
838 }
839
840 for _, scope := range []string{"", "any-session", "some-other"} {
841 f.SetActiveSession(scope)
842 got := snapshotIDs(f.SnapshotAll())
843 if len(got) != 1 || got[0] != a.ID {
844 t.Errorf("scope=%q: ids = %v; want legacy agent visible", scope, got)
845 }
846 }
847
848 _ = f.Stop(a.ID)
849 a.Wait()
850}
851
852func snapshotIDs(ss []AgentSnapshot) []string {
853 out := make([]string, len(ss))
854 for i, s := range ss {
855 out[i] = s.ID
856 }
857 return out
858}