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// in.sock unix socket inbox (child-owned)
155// session.json persistent agent session (child-owned)
156// meta.json static metadata (id, task)
157func (f *Swarm) agentStateDir(id string) string {
158 return filepath.Join(f.cfg.Root, "agents", id)
159}
160
161// SpawnRequest configures a Spawn. Only Task is required; the rest
162// are optional. Model + Provider, when set, get baked into the
163// child argv as --model / --provider so the agent runs against the
164// chosen model regardless of the parent's current selection.
165type SpawnRequest struct {
166 Task string
167 Model string // optional override; child resolves default if empty
168 Provider string // optional override; usually paired with Model
169}
170
171// Spawn creates a new Agent for the given task, allocates its
172// on-disk state directory (events log, inbox socket path, session
173// file path), and starts the Runner on a background goroutine. The
174// returned Agent is already in StatusRunning (or StatusFailed if
175// state setup failed before the goroutine started). This is the
176// historical signature; callers that want to override the child's
177// model use SpawnReq instead.
178func (f *Swarm) Spawn(ctx context.Context, task string) (*Agent, error) {
179 return f.SpawnReq(ctx, SpawnRequest{Task: task})
180}
181
182// SpawnReq is the full-fat variant of Spawn that accepts a
183// SpawnRequest. Existing callers can keep using Spawn; new code that
184// wants to pin the child's model uses this.
185//
186// Every spawned agent runs with cwd == cfg.RepoRoot — the same
187// working directory as the host. No per-agent worktree, no branch,
188// no isolation. The user explicitly opted out of the worktree flow.
189func (f *Swarm) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
190 task := strings.TrimSpace(req.Task)
191 if task == "" {
192 return nil, errors.New("swarm: empty task")
193 }
194 id := newAgentID(task, f.cfg.Now())
195 dir := f.cfg.RepoRoot
196
197 stateDir := f.agentStateDir(id)
198 if err := os.MkdirAll(stateDir, 0o755); err != nil {
199 return nil, fmt.Errorf("swarm state dir: %w", err)
200 }
201 logPath := filepath.Join(stateDir, "events.jsonl")
202 sessionPath := filepath.Join(stateDir, "session.json")
203 // Unix sockets have a hard 104-byte path limit on darwin and 108
204 // on linux. Long ZOT_HOME paths plus an agent slug blow that cap
205 // quickly. Pick the shortest path that still keeps sockets
206 // per-swarm-root so two zot instances on the same machine don't
207 // collide. inboxSocketPath falls back from $TMPDIR to /tmp if
208 // neither is short enough.
209 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
210 if err != nil {
211 return nil, fmt.Errorf("swarm inbox path: %w", err)
212 }
213
214 // Snapshot activeSession under the lock; we use it twice (struct
215 // init below + persistence). Reading it without the lock would
216 // race a concurrent SetActiveSession call.
217 f.mu.Lock()
218 sessionID := f.activeSession
219 f.mu.Unlock()
220
221 a := &Agent{
222 ID: id,
223 Task: task,
224 Dir: dir,
225 Started: f.cfg.Now(),
226 Model: strings.TrimSpace(req.Model),
227 Provider: strings.TrimSpace(req.Provider),
228 SessionID: sessionID,
229 InboxPath: inboxPath,
230 EventLogPath: logPath,
231 SessionPath: sessionPath,
232 inbox: NewInbox(inboxPath),
233 status: StatusPending,
234 activity: "queued",
235 done: make(chan struct{}),
236 }
237 a.ctx, a.cancel = context.WithCancel(ctx)
238 a.runner = f.cfg.NewRunner(a)
239
240 f.mu.Lock()
241 f.agents[id] = a
242 f.order = append(f.order, id)
243 f.mu.Unlock()
244
245 // Persist the agent's identity so a later `zot` invocation can
246 // reload it from disk via Swarm.Reload. Best-effort: if the disk
247 // is read-only we still let the runner start, the user just won't
248 // see this agent on the next launch.
249 _ = writeAgentMeta(stateDir, a)
250
251 go f.run(a)
252 return a, nil
253}
254
255// SendInput delivers msg to the agent's inbox. msg should already
256// include the leading kind ("user " + text, "cancel", "shutdown").
257// Returns an error wrapping ErrNotReady if the child hasn't opened
258// its listener yet; the caller can retry or surface the wait.
259func (f *Swarm) SendInput(id, msg string) error {
260 a := f.Get(id)
261 if a == nil {
262 return fmt.Errorf("no such agent %q", id)
263 }
264 if a.inbox == nil {
265 return fmt.Errorf("agent %s has no inbox", a.ID)
266 }
267 return a.inbox.SendInput(msg)
268}
269
270// SendUserTurn is sugar for the common "send the next user turn"
271// case. It quotes nothing and forwards verbatim; callers are
272// expected to have already trimmed and expanded the text.
273func (f *Swarm) SendUserTurn(id, text string) error {
274 return f.SendInput(id, "user "+text)
275}
276
277func (f *Swarm) run(a *Agent) {
278 a.setStatus(StatusRunning)
279 a.setActivity("starting")
280 err := a.runner.Run(a.ctx, agentSink{a: a})
281 a.mu.Lock()
282 a.finished = f.cfg.Now()
283 switch {
284 case a.status == StatusKilled:
285 // Already finalised by Stop.
286 case errors.Is(err, context.Canceled):
287 a.status = StatusKilled
288 a.activity = "cancelled"
289 case err != nil:
290 a.status = StatusFailed
291 a.activity = "error: " + truncate(err.Error(), 120)
292 a.lastErr = err
293 default:
294 a.status = StatusDone
295 a.activity = "done"
296 }
297 a.mu.Unlock()
298 close(a.done)
299}
300
301// List returns a snapshot of every agent in creation order. The
302// returned slice is a copy; callers may iterate without holding the
303// swarm lock. Agent fields are read under their own mutex during
304// formatting in Snapshot.
305func (f *Swarm) List() []*Agent {
306 f.mu.Lock()
307 defer f.mu.Unlock()
308 out := make([]*Agent, 0, len(f.order))
309 for _, id := range f.order {
310 out = append(out, f.agents[id])
311 }
312 return out
313}
314
315// Get returns the agent with the given (possibly truncated) id, or
316// nil. Matching is prefix-based so the user can type the first few
317// characters of a long id.
318func (f *Swarm) Get(id string) *Agent {
319 f.mu.Lock()
320 defer f.mu.Unlock()
321 if a, ok := f.agents[id]; ok {
322 return a
323 }
324 // Prefix match.
325 var hits []*Agent
326 for k, a := range f.agents {
327 if strings.HasPrefix(k, id) {
328 hits = append(hits, a)
329 }
330 }
331 if len(hits) == 1 {
332 return hits[0]
333 }
334 return nil
335}
336
337// Stop cancels the agent's context. The Runner should observe the
338// cancellation and return promptly; the goroutine then finalises the
339// agent in StatusKilled. Also closes the supervisor-side inbox handle
340// so any pending SendInput retries fail fast instead of dialing a
341// socket about to be unlinked.
342//
343// Stop is a no-op for any agent that's not in a live runnable state
344// — Done / Failed / Killed (already finalised) and Detached (no
345// in-process runner; reloaded from disk). Calling Stop on a detached
346// agent must not crash: buildDetachedAgent doesn't allocate a
347// context/cancel pair because there's nothing to cancel.
348func (f *Swarm) Stop(id string) error {
349 a := f.Get(id)
350 if a == nil {
351 return fmt.Errorf("no such agent %q", id)
352 }
353 a.mu.Lock()
354 switch a.status {
355 case StatusDone, StatusFailed, StatusKilled, StatusDetached:
356 a.mu.Unlock()
357 return nil
358 }
359 a.status = StatusKilled
360 a.activity = "stopped"
361 a.mu.Unlock()
362 // Belt-and-braces guard against the cancel-less case (detached
363 // agents skip this branch above, but a future code path that
364 // builds an Agent without a runner shouldn't crash the
365 // supervisor).
366 if a.cancel != nil {
367 a.cancel()
368 }
369 if a.inbox != nil {
370 _ = a.inbox.Close()
371 }
372 return nil
373}
374
375// StopAll cancels every running agent. Used on shutdown.
376func (f *Swarm) StopAll() {
377 for _, a := range f.List() {
378 _ = f.Stop(a.ID)
379 }
380}
381
382// Remove tears down the per-agent state for a terminated agent. It
383// is an error to remove an agent that's still running; call Stop
384// first and wait for the status to settle. Detached agents
385// (reloaded from disk) remove cleanly because they have no live
386// runner racing for the same files.
387//
388// Agents share the host's working tree, so Remove never touches
389// any source file — it only deletes the agent's state directory
390// under <root>/agents/<id>/.
391func (f *Swarm) Remove(id string) error {
392 a := f.Get(id)
393 if a == nil {
394 return fmt.Errorf("no such agent %q", id)
395 }
396 a.mu.Lock()
397 st := a.status
398 a.mu.Unlock()
399 if st == StatusRunning || st == StatusPending {
400 return fmt.Errorf("agent %s still %s", a.ID, st)
401 }
402 // Best-effort cleanup of the per-agent state directory
403 // (meta.json, events.jsonl, session.json, in.sock if it's
404 // local). Failing here would leave the user with no recourse,
405 // so swallow the error.
406 _ = os.RemoveAll(f.agentStateDir(a.ID))
407 f.mu.Lock()
408 delete(f.agents, a.ID)
409 for i, k := range f.order {
410 if k == a.ID {
411 f.order = append(f.order[:i], f.order[i+1:]...)
412 break
413 }
414 }
415 f.mu.Unlock()
416 return nil
417}
418
419// Snapshot returns a read-only view of one agent. Safe for the TUI
420// goroutine to call repeatedly; never blocks on the Runner.
421type AgentSnapshot struct {
422 ID string
423 Task string
424 Dir string
425 Status Status
426 Activity string
427 Started time.Time
428 Finished time.Time
429 Err string
430 Tail string // last few transcript lines, joined with "\n"
431 Lines []string // full transcript (already capped by Agent.appendTranscript)
432
433 // Model and Provider expose the per-agent overrides set at
434 // Spawn time (empty when the agent inherits the child's default
435 // resolution). The dashboard surfaces these so the user can
436 // confirm which model an agent is running against.
437 Model string
438 Provider string
439
440 // Paths to the agent's durable state. Surface them in the
441 // snapshot so the dashboard / /swarm open can read events.jsonl
442 // or resume the session without going back through the Agent.
443 InboxPath string
444 EventLogPath string
445 SessionPath string
446}
447
448// Snapshot copies the live agent state into a value the caller can
449// inspect at leisure.
450func (a *Agent) Snapshot() AgentSnapshot {
451 a.mu.Lock()
452 defer a.mu.Unlock()
453 tail := strings.Join(lastN(a.transcript, 6), "\n")
454 lines := make([]string, len(a.transcript))
455 copy(lines, a.transcript)
456 errStr := ""
457 if a.lastErr != nil {
458 errStr = a.lastErr.Error()
459 }
460 return AgentSnapshot{
461 ID: a.ID, Task: a.Task, Dir: a.Dir,
462 Status: a.status, Activity: a.activity,
463 Started: a.Started, Finished: a.finished,
464 Err: errStr, Tail: tail, Lines: lines,
465 Model: a.Model,
466 Provider: a.Provider,
467 InboxPath: a.InboxPath,
468 EventLogPath: a.EventLogPath,
469 SessionPath: a.SessionPath,
470 }
471}
472
473// SnapshotAll returns snapshots of every agent in creation order,
474// scoped to the active session when one is set.
475//
476// Scoping rules:
477// - activeSession == "": no filter, every agent is returned
478// (historical behaviour; used by tests and scripted callers).
479// - activeSession != "": include only agents whose SessionID
480// matches activeSession OR is empty. The empty-id pass-through
481// keeps pre-upgrade agents (their meta.json was written before
482// session_id existed) visible from any session so the user
483// doesn't lose access after the schema bump.
484func (f *Swarm) SnapshotAll() []AgentSnapshot {
485 agents := f.List()
486 f.mu.Lock()
487 active := f.activeSession
488 f.mu.Unlock()
489
490 out := make([]AgentSnapshot, 0, len(agents))
491 for _, a := range agents {
492 if active != "" && a.SessionID != "" && a.SessionID != active {
493 continue
494 }
495 out = append(out, a.Snapshot())
496 }
497 // Sort by start time for a stable, deterministic listing.
498 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
499 return out
500}
501
502// agentSink is the Sink the Swarm hands to each Runner.
503type agentSink struct{ a *Agent }
504
505func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
506func (s agentSink) Transcript(chunk string) { s.a.appendTranscript(chunk) }
507
508func truncate(s string, n int) string {
509 if len(s) <= n {
510 return s
511 }
512 if n <= 3 {
513 return strings.Repeat(".", n)
514 }
515 return s[:n-3] + "..."
516}
517
518func lastN(lines []string, n int) []string {
519 if len(lines) <= n {
520 return lines
521 }
522 return lines[len(lines)-n:]
523}