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