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