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