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