1// Package swarm implements zot's multi-agent supervisor.
2//
3// A Swarm manages a set of headless zot subprocesses ("agents")
4// that share the host's working directory. The interactive TUI
5// exposes the supervisor through the /swarm slash command and a
6// dashboard dialog; non-TUI code can drive it directly through
7// this package.
8//
9// Every agent runs with cwd == the parent zot's RepoRoot — the
10// same files the user sees, the same files the main agent edits.
11// There is no git worktree, no per-agent branch, no isolation. If
12// you want parallel edits on a separate branch, use normal git
13// tooling (a real worktree, a different terminal) yourself.
14//
15// Each Agent has:
16// - a unique id (short slug + nanoseconds)
17// - a Runner (the thing that actually executes the task)
18// - a Status string + Activity string that the dashboard reads
19//
20// The Runner abstraction means tests can swap a fake in instead of
21// really spawning a subprocess; the production Runner shells out to
22// `zot --swarm-agent ...` so we reuse zot's own model resolution
23// and tooling without re-implementing the agent loop.
24package swarm
25
26import (
27 "context"
28 "errors"
29 "fmt"
30 "os"
31 "path/filepath"
32 "sort"
33 "strings"
34 "sync"
35 "time"
36)
37
38// Status is the high-level lifecycle state of an Agent.
39type Status string
40
41const (
42 StatusPending Status = "pending" // created, not started yet
43 StatusRunning Status = "running" // Runner.Run is in flight
44 StatusDone Status = "done" // Runner.Run returned nil
45 StatusFailed Status = "failed" // Runner.Run returned an error
46 StatusKilled Status = "killed" // Stop() called before completion
47 StatusDetached Status = "detached" // reloaded from disk; no live runner
48)
49
50// Config configures a Swarm.
51type Config struct {
52 // Root is the directory under which per-agent state files live.
53 // Typically <ZotHome>/swarm, but tests pass a tempdir.
54 Root string
55
56 // RepoRoot is the working directory every spawned agent runs
57 // in — the same cwd the parent zot is using. There is no
58 // per-agent isolation: agents edit the host's files directly.
59 RepoRoot string
60
61 // NewRunner produces the Runner for an Agent. If nil, the default
62 // `zot --swarm-agent ...` exec runner is used. Tests inject a fake
63 // here.
64 NewRunner func(a *Agent) Runner
65
66 // Now is a clock seam for tests; defaults to time.Now.
67 Now func() time.Time
68}
69
70// Runner executes one agent task. Run blocks until the task finishes,
71// is cancelled via ctx, or hits an unrecoverable error.
72//
73// Run should report progress by writing short human-readable strings
74// to the activity channel and final transcript text to transcript.
75// Both channels are non-blocking sinks owned by the Swarm; if the
76// dashboard isn't reading, sends are dropped.
77type Runner interface {
78 Run(ctx context.Context, sink Sink) error
79}
80
81// Sink is how a Runner reports activity and transcript back to the
82// supervisor. All methods are safe to call from any goroutine and
83// never block.
84type Sink interface {
85 // Activity sets the one-line "what is this agent doing right now"
86 // string shown in the dashboard.
87 Activity(msg string)
88 // Transcript appends a chunk of agent output (typically a final
89 // assistant message) to the agent's running transcript.
90 Transcript(chunk string)
91}
92
93// Swarm supervises a set of Agents.
94type Swarm struct {
95 cfg Config
96
97 mu sync.Mutex
98 agents map[string]*Agent
99 order []string // creation order for stable listing
100
101 // activeSession is the host session id the dashboard is
102 // currently scoped to. When non-empty, SnapshotAll filters out
103 // agents whose SessionID doesn't match (legacy / unscoped
104 // agents with SessionID == "" are always shown). Spawn stamps
105 // new agents with this value so they appear only in the session
106 // that created them. When empty (the default), the historical
107 // "show everything" behaviour is preserved — important for
108 // tests and any scripted use of the Swarm that doesn't bother
109 // with sessions.
110 activeSession string
111}
112
113// New constructs a Swarm from cfg. Missing config fields are filled
114// with defaults.
115func New(cfg Config) *Swarm {
116 if cfg.Now == nil {
117 cfg.Now = time.Now
118 }
119 if cfg.NewRunner == nil {
120 cfg.NewRunner = func(a *Agent) Runner { return &execRunner{agent: a} }
121 }
122 return &Swarm{
123 cfg: cfg,
124 agents: map[string]*Agent{},
125 }
126}
127
128// SetActiveSession scopes the dashboard view (and Spawn stamping)
129// to a particular host zot session id. Pass empty to clear the
130// scope and revert to "show every agent" (the original behaviour).
131//
132// Existing in-memory agents keep their SessionID; only the filter
133// applied at snapshot time changes. So swapping the active session
134// with /sessions instantly re-narrows the dashboard without
135// touching any agent state.
136func (f *Swarm) SetActiveSession(id string) {
137 f.mu.Lock()
138 defer f.mu.Unlock()
139 f.activeSession = id
140}
141
142// ActiveSession returns the current scope, mostly for tests and
143// diagnostics. Empty means "no scope; show everything".
144func (f *Swarm) ActiveSession() string {
145 f.mu.Lock()
146 defer f.mu.Unlock()
147 return f.activeSession
148}
149
150// agentStateDir is the per-agent state directory laid out as:
151//
152// <root>/agents/<id>/
153// events.jsonl durable event log (runner-owned)
154// session.json persistent agent session (child-owned)
155// meta.json static metadata (id, task)
156//
157// The transient unix socket inbox lives in a probed runtime directory,
158// which may be on a different filesystem from this durable state.
159func (f *Swarm) agentStateDir(id string) string {
160 return filepath.Join(f.cfg.Root, "agents", id)
161}
162
163// SpawnRequest configures a Spawn. Only Task is required; the rest
164// are optional. Model + Provider, when set, get baked into the
165// child argv as --model / --provider so the agent runs against the
166// chosen model regardless of the parent's current selection.
167type SpawnRequest struct {
168 Task string
169 Model string // optional override; child resolves default if empty
170 Provider string // optional override; usually paired with Model
171}
172
173// Spawn creates a new Agent for the given task, allocates its
174// on-disk state directory (events log, inbox socket path, session
175// file path), and starts the Runner on a background goroutine. The
176// returned Agent is already in StatusRunning (or StatusFailed if
177// state setup failed before the goroutine started). This is the
178// historical signature; callers that want to override the child's
179// model use SpawnReq instead.
180func (f *Swarm) Spawn(ctx context.Context, task string) (*Agent, error) {
181 return f.SpawnReq(ctx, SpawnRequest{Task: task})
182}
183
184// SpawnReq is the full-fat variant of Spawn that accepts a
185// SpawnRequest. Existing callers can keep using Spawn; new code that
186// wants to pin the child's model uses this.
187//
188// Every spawned agent runs with cwd == cfg.RepoRoot — the same
189// working directory as the host. No per-agent worktree, no branch,
190// no isolation. The user explicitly opted out of the worktree flow.
191func (f *Swarm) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
192 task := strings.TrimSpace(req.Task)
193 if task == "" {
194 return nil, errors.New("swarm: empty task")
195 }
196 id := newAgentID(task, f.cfg.Now())
197 dir := f.cfg.RepoRoot
198
199 stateDir := f.agentStateDir(id)
200 if err := os.MkdirAll(stateDir, 0o755); err != nil {
201 return nil, fmt.Errorf("swarm state dir: %w", err)
202 }
203 logPath := filepath.Join(stateDir, "events.jsonl")
204 sessionPath := filepath.Join(stateDir, "session.json")
205 // Keep the transient unix socket outside the durable state root.
206 // ZOT_HOME may live on a shared or network filesystem that supports
207 // regular state files but cannot host unix socket nodes.
208 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
209 if err != nil {
210 return nil, fmt.Errorf("swarm inbox path: %w", err)
211 }
212
213 // Snapshot activeSession under the lock; we use it twice (struct
214 // init below + persistence). Reading it without the lock would
215 // race a concurrent SetActiveSession call.
216 f.mu.Lock()
217 sessionID := f.activeSession
218 f.mu.Unlock()
219
220 a := &Agent{
221 ID: id,
222 Task: task,
223 Dir: dir,
224 Started: f.cfg.Now(),
225 Model: strings.TrimSpace(req.Model),
226 Provider: strings.TrimSpace(req.Provider),
227 SessionID: sessionID,
228 InboxPath: inboxPath,
229 EventLogPath: logPath,
230 SessionPath: sessionPath,
231 inbox: NewInbox(inboxPath),
232 status: StatusPending,
233 activity: "queued",
234 done: make(chan struct{}),
235 }
236 a.ctx, a.cancel = context.WithCancel(ctx)
237 a.runner = f.cfg.NewRunner(a)
238
239 f.mu.Lock()
240 f.agents[id] = a
241 f.order = append(f.order, id)
242 f.mu.Unlock()
243
244 // Persist the agent's identity so a later `zot` invocation can
245 // reload it from disk via Swarm.Reload. Best-effort: if the disk
246 // is read-only we still let the runner start, the user just won't
247 // see this agent on the next launch.
248 _ = writeAgentMeta(stateDir, a)
249
250 go f.run(a)
251 return a, nil
252}
253
254// SendInput delivers msg to the agent's inbox. msg should already
255// include the leading kind ("user " + text, "cancel", "shutdown").
256// Returns an error wrapping ErrNotReady if the child hasn't opened
257// its listener yet; the caller can retry or surface the wait.
258func (f *Swarm) SendInput(id, msg string) error {
259 a := f.Get(id)
260 if a == nil {
261 return fmt.Errorf("no such agent %q", id)
262 }
263 if a.inbox == nil {
264 return fmt.Errorf("agent %s has no inbox", a.ID)
265 }
266 return a.inbox.SendInput(msg)
267}
268
269// SendUserTurn is sugar for the common "send the next user turn"
270// case. It quotes nothing and forwards verbatim; callers are
271// expected to have already trimmed and expanded the text.
272func (f *Swarm) SendUserTurn(id, text string) error {
273 return f.SendInput(id, "user "+text)
274}
275
276func (f *Swarm) run(a *Agent) {
277 a.setStatus(StatusRunning)
278 a.setActivity("starting")
279 err := a.runner.Run(a.ctx, agentSink{a: a})
280 a.mu.Lock()
281 a.finished = f.cfg.Now()
282 switch {
283 case a.status == StatusKilled:
284 // Already finalised by Stop.
285 case errors.Is(err, context.Canceled):
286 a.status = StatusKilled
287 a.activity = "cancelled"
288 case err != nil:
289 a.status = StatusFailed
290 a.activity = "error: " + truncate(err.Error(), 120)
291 a.lastErr = err
292 default:
293 a.status = StatusDone
294 a.activity = "done"
295 }
296 a.mu.Unlock()
297 close(a.done)
298}
299
300// List returns a snapshot of every agent in creation order. The
301// returned slice is a copy; callers may iterate without holding the
302// swarm lock. Agent fields are read under their own mutex during
303// formatting in Snapshot.
304func (f *Swarm) List() []*Agent {
305 f.mu.Lock()
306 defer f.mu.Unlock()
307 out := make([]*Agent, 0, len(f.order))
308 for _, id := range f.order {
309 out = append(out, f.agents[id])
310 }
311 return out
312}
313
314// Get returns the agent with the given (possibly truncated) id, or
315// nil. Matching is prefix-based so the user can type the first few
316// characters of a long id.
317func (f *Swarm) Get(id string) *Agent {
318 f.mu.Lock()
319 defer f.mu.Unlock()
320 if a, ok := f.agents[id]; ok {
321 return a
322 }
323 // Prefix match.
324 var hits []*Agent
325 for k, a := range f.agents {
326 if strings.HasPrefix(k, id) {
327 hits = append(hits, a)
328 }
329 }
330 if len(hits) == 1 {
331 return hits[0]
332 }
333 return nil
334}
335
336// Stop cancels the agent's context. The Runner should observe the
337// cancellation and return promptly; the goroutine then finalises the
338// agent in StatusKilled. Also closes the supervisor-side inbox handle
339// so any pending SendInput retries fail fast instead of dialing a
340// socket about to be unlinked.
341//
342// Stop is a no-op for any agent that's not in a live runnable state
343// — Done / Failed / Killed (already finalised) and Detached (no
344// in-process runner; reloaded from disk). Calling Stop on a detached
345// agent must not crash: buildDetachedAgent doesn't allocate a
346// context/cancel pair because there's nothing to cancel.
347func (f *Swarm) Stop(id string) error {
348 a := f.Get(id)
349 if a == nil {
350 return fmt.Errorf("no such agent %q", id)
351 }
352 a.mu.Lock()
353 switch a.status {
354 case StatusDone, StatusFailed, StatusKilled, StatusDetached:
355 a.mu.Unlock()
356 return nil
357 }
358 a.status = StatusKilled
359 a.activity = "stopped"
360 a.mu.Unlock()
361 // Belt-and-braces guard against the cancel-less case (detached
362 // agents skip this branch above, but a future code path that
363 // builds an Agent without a runner shouldn't crash the
364 // supervisor).
365 if a.cancel != nil {
366 a.cancel()
367 }
368 if a.inbox != nil {
369 _ = a.inbox.Close()
370 }
371 return nil
372}
373
374// StopAll cancels every running agent. Used on shutdown.
375func (f *Swarm) StopAll() {
376 for _, a := range f.List() {
377 _ = f.Stop(a.ID)
378 }
379}
380
381// Remove tears down the per-agent state for a terminated agent. It
382// is an error to remove an agent that's still running; call Stop
383// first and wait for the status to settle. Detached agents
384// (reloaded from disk) remove cleanly because they have no live
385// runner racing for the same files.
386//
387// Agents share the host's working tree, so Remove never touches
388// any source file — it only deletes the agent's state directory
389// under <root>/agents/<id>/.
390func (f *Swarm) Remove(id string) error {
391 a := f.Get(id)
392 if a == nil {
393 return fmt.Errorf("no such agent %q", id)
394 }
395 a.mu.Lock()
396 st := a.status
397 a.mu.Unlock()
398 if st == StatusRunning || st == StatusPending {
399 return fmt.Errorf("agent %s still %s", a.ID, st)
400 }
401 // Best-effort cleanup of the per-agent state directory
402 // (meta.json, events.jsonl, session.json). Failing here would leave
403 // the user with no recourse,
404 // so swallow the error.
405 _ = os.RemoveAll(f.agentStateDir(a.ID))
406 f.mu.Lock()
407 delete(f.agents, a.ID)
408 for i, k := range f.order {
409 if k == a.ID {
410 f.order = append(f.order[:i], f.order[i+1:]...)
411 break
412 }
413 }
414 f.mu.Unlock()
415 return nil
416}
417
418// Snapshot returns a read-only view of one agent. Safe for the TUI
419// goroutine to call repeatedly; never blocks on the Runner.
420type AgentSnapshot struct {
421 ID string
422 Task string
423 Dir string
424 Status Status
425 Activity string
426 Started time.Time
427 Finished time.Time
428 Err string
429 Tail string // last few transcript lines, joined with "\n"
430 Lines []string // full transcript (already capped by Agent.appendTranscript)
431
432 // Model and Provider expose the per-agent overrides set at
433 // Spawn time (empty when the agent inherits the child's default
434 // resolution). The dashboard surfaces these so the user can
435 // confirm which model an agent is running against.
436 Model string
437 Provider string
438
439 // Paths to the agent's durable state. Surface them in the
440 // snapshot so the dashboard / /swarm open can read events.jsonl
441 // or resume the session without going back through the Agent.
442 InboxPath string
443 EventLogPath string
444 SessionPath string
445}
446
447// Snapshot copies the live agent state into a value the caller can
448// inspect at leisure.
449func (a *Agent) Snapshot() AgentSnapshot {
450 a.mu.Lock()
451 defer a.mu.Unlock()
452 tail := strings.Join(lastN(a.transcript, 6), "\n")
453 lines := make([]string, len(a.transcript))
454 copy(lines, a.transcript)
455 errStr := ""
456 if a.lastErr != nil {
457 errStr = a.lastErr.Error()
458 }
459 return AgentSnapshot{
460 ID: a.ID, Task: a.Task, Dir: a.Dir,
461 Status: a.status, Activity: a.activity,
462 Started: a.Started, Finished: a.finished,
463 Err: errStr, Tail: tail, Lines: lines,
464 Model: a.Model,
465 Provider: a.Provider,
466 InboxPath: a.InboxPath,
467 EventLogPath: a.EventLogPath,
468 SessionPath: a.SessionPath,
469 }
470}
471
472// SnapshotAll returns snapshots of every agent in creation order,
473// scoped to the active session when one is set.
474//
475// Scoping rules:
476// - activeSession == "": no filter, every agent is returned
477// (historical behaviour; used by tests and scripted callers).
478// - activeSession != "": include only agents whose SessionID
479// matches activeSession OR is empty. The empty-id pass-through
480// keeps pre-upgrade agents (their meta.json was written before
481// session_id existed) visible from any session so the user
482// doesn't lose access after the schema bump.
483func (f *Swarm) SnapshotAll() []AgentSnapshot {
484 agents := f.List()
485 f.mu.Lock()
486 active := f.activeSession
487 f.mu.Unlock()
488
489 out := make([]AgentSnapshot, 0, len(agents))
490 for _, a := range agents {
491 if active != "" && a.SessionID != "" && a.SessionID != active {
492 continue
493 }
494 out = append(out, a.Snapshot())
495 }
496 // Sort by start time for a stable, deterministic listing.
497 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
498 return out
499}
500
501// agentSink is the Sink the Swarm hands to each Runner.
502type agentSink struct{ a *Agent }
503
504func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
505func (s agentSink) Transcript(chunk string) { s.a.appendTranscript(chunk) }
506
507func truncate(s string, n int) string {
508 if len(s) <= n {
509 return s
510 }
511 if n <= 3 {
512 return strings.Repeat(".", n)
513 }
514 return s[:n-3] + "..."
515}
516
517func lastN(lines []string, n int) []string {
518 if len(lines) <= n {
519 return lines
520 }
521 return lines[len(lines)-n:]
522}