1package swarm
2
3import (
4 "bufio"
5 "context"
6 "encoding/json"
7 "fmt"
8 "io"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strings"
13 "time"
14)
15
16// execRunner spawns `zot --swarm-agent <inbox> --session <path>` in
17// the host's working directory (Agent.Dir, which is always the parent
18// zot's RepoRoot) and consumes its JSONL event stream on stdout.
19//
20// Why a long-lived daemon and not `zot --print`: the supervisor and
21// the user expect agents to keep accepting follow-up prompts. A
22// one-shot subprocess can't do that; this design gives each swarm
23// agent a persistent session file plus an inbox socket the parent
24// writes to, mirroring Claude Code's "Agents view" model.
25//
26// Events flow:
27//
28// child stdout --> decoder --> EventLog (events.jsonl)
29// --> Sink (Activity/Transcript)
30//
31// The on-disk log is the durable record. The Sink updates are an
32// in-memory mirror so the dashboard doesn't have to tail the file
33// for the parent's own agents. /swarm open in a separate zot would
34// read the log directly.
35type execRunner struct {
36 agent *Agent
37
38 // Command overrides the default `zot --swarm-agent ...`
39 // invocation. Tests set this to a fake binary (or `go run`
40 // against a tiny stub program) so the supervisor logic can be
41 // tested without a real child. Production code leaves it nil.
42 Command []string
43
44 // SessionPath is the agent's session file. Empty means "defer
45 // to r.agent.SessionPath", which Swarm.Spawn always populates
46 // with <swarm-root>/agents/<id>/session.json. Tests that
47 // hand-build an Agent without going through Spawn must set
48 // one of the two; the runner refuses to invent a fallback
49 // because the only plausible one (<Dir>/.zot/session.json)
50 // would litter the user's repo — every agent's Dir points
51 // at it directly.
52 SessionPath string
53}
54
55// swarmAgentArgsOpts captures every dynamic input to swarmAgentArgs
56// so future per-agent overrides (e.g. tools, reasoning) can be added
57// without churning the signature. The fields map 1:1 onto child CLI
58// flags; empty values omit the flag entirely and let the child
59// resolve a default the same way a normal `zot` invocation does.
60type swarmAgentArgsOpts struct {
61 Exe string
62 Dir string
63 SessionPath string
64 InboxPath string
65 Task string
66 Model string
67 Provider string
68}
69
70// defaultChildArgs builds the argv execRunner uses when its Command
71// override is empty. Centralised so the spawn-vs-resume decision
72// (whether to pass the original Task as a positional) lives in one
73// place that tests can hit directly without going through Run's
74// side effects.
75//
76// On Spawn (Resuming==false) we pass the task so the child's first
77// turn runs immediately. On Resume (Resuming==true) we omit it: the
78// child reopens the existing session file, loads the prior
79// conversation, and just waits on the inbox for the next prompt.
80// Re-firing the task on every resume produces a duplicate turn that
81// collides with whatever the user types next, surfacing the
82// "agent busy; send 'cancel' first" error between two assistant
83// messages — which is exactly the bug this helper fixes.
84func defaultChildArgs(exe string, a *Agent, sessionPath, inboxPath string) []string {
85 task := a.Task
86 if a.Resuming {
87 task = ""
88 }
89 return swarmAgentArgs(swarmAgentArgsOpts{
90 Exe: exe,
91 Dir: a.Dir,
92 SessionPath: sessionPath,
93 InboxPath: inboxPath,
94 Task: task,
95 Model: a.Model,
96 Provider: a.Provider,
97 })
98}
99
100// swarmAgentArgs builds the argv used when execRunner.Command is
101// empty. Pulled out so tests can lock in the flag set without
102// actually spawning a subprocess.
103func swarmAgentArgs(opts swarmAgentArgsOpts) []string {
104 args := []string{
105 opts.Exe,
106 "--swarm-agent", opts.InboxPath,
107 "--session", opts.SessionPath,
108 "--cwd", opts.Dir,
109 }
110 if opts.Model != "" {
111 args = append(args, "--model", opts.Model)
112 }
113 if opts.Provider != "" {
114 args = append(args, "--provider", opts.Provider)
115 }
116 if opts.Task != "" {
117 // First task is positional so the child treats it as the
118 // initial user turn; subsequent turns arrive via the inbox.
119 args = append(args, opts.Task)
120 }
121 return args
122}
123
124func (r *execRunner) Run(ctx context.Context, sink Sink) error {
125 // SessionPath resolution order:
126 // 1. explicit r.SessionPath set by the test / caller
127 // 2. r.agent.SessionPath baked in by Swarm.Spawn — the
128 // production path. Always lives under
129 // <swarm-root>/agents/<id>/session.json so the per-
130 // agent state is entirely outside the working tree.
131 // Crucial because Agent.Dir points at the user's repo;
132 // any .zot/ scratch directory under Dir would litter
133 // their source tree.
134 //
135 // There is no third fallback. If neither path is set we
136 // refuse to start instead of inventing a directory; that
137 // way a misconfigured caller fails loudly the first time
138 // instead of silently dumping session data into someone's
139 // repo.
140 sessionPath := r.SessionPath
141 if sessionPath == "" {
142 sessionPath = r.agent.SessionPath
143 }
144 if sessionPath == "" {
145 return fmt.Errorf("swarm: agent missing session path (set SpawnRequest via Swarm.SpawnReq, or hand-build Agent with SessionPath populated)")
146 }
147 if err := os.MkdirAll(filepath.Dir(sessionPath), 0o755); err != nil {
148 return fmt.Errorf("session dir: %w", err)
149 }
150
151 inboxPath := r.agent.InboxPath
152 logPath := r.agent.EventLogPath
153 if logPath == "" {
154 return fmt.Errorf("swarm: agent missing event log path")
155 }
156 log, err := OpenEventLog(logPath)
157 if err != nil {
158 return err
159 }
160 defer log.Close()
161
162 args := r.Command
163 if len(args) == 0 {
164 exe, err := os.Executable()
165 if err != nil {
166 return fmt.Errorf("locate self: %w", err)
167 }
168 args = defaultChildArgs(exe, r.agent, sessionPath, inboxPath)
169 }
170
171 cmd := exec.CommandContext(ctx, args[0], args[1:]...)
172 cmd.Dir = r.agent.Dir
173 cmd.Env = append(os.Environ(),
174 "ZOT_SWARM_AGENT_ID="+r.agent.ID,
175 "ZOT_SWARM_EVENT_LOG="+logPath,
176 )
177
178 stdout, err := cmd.StdoutPipe()
179 if err != nil {
180 return err
181 }
182 stderr, err := cmd.StderrPipe()
183 if err != nil {
184 return err
185 }
186
187 // "spawning" is briefly shown until the first event arrives;
188 // the child's "spawned" lifecycle event then overwrites it.
189 sink.Activity("starting")
190 if err := cmd.Start(); err != nil {
191 return err
192 }
193
194 // stdout: parsed as JSONL. Every well-formed event is appended
195 // to the durable log AND forwarded to the in-memory sink so the
196 // dashboard updates without having to tail the file. Malformed
197 // lines are surfaced as plain transcript so they don't vanish.
198 done := make(chan struct{}, 2)
199 go func() {
200 defer func() { done <- struct{}{} }()
201 dec := bufio.NewReader(stdout)
202 for {
203 line, err := dec.ReadBytes('\n')
204 if len(line) > 0 {
205 trimmed := strings.TrimRight(string(line), "\r\n")
206 if trimmed == "" {
207 goto next
208 }
209 if ev, ok := parseEventLine(trimmed); ok {
210 _ = log.Append(ev)
211 applyEventToSink(ev, sink)
212 // Fan turn_end up to any subscriber on the
213 // supervised Agent. Daemons stay alive across
214 // many turns, so Wait()-style hooks would
215 // never fire; per-turn callbacks let auto-
216 // swarm summarise as each task completes.
217 if ev.Type == "turn_end" && r.agent != nil {
218 r.agent.mu.Lock()
219 fn := r.agent.OnTurnEnd
220 r.agent.mu.Unlock()
221 if fn != nil {
222 step, _ := ev.Data["step"].(float64)
223 errMsg, _ := ev.Data["error"].(string)
224 go fn(int(step), errMsg)
225 }
226 }
227 } else {
228 // Non-JSON output. Keep it as transcript so an
229 // accidental fmt.Println in the child still
230 // shows up in the dashboard, and log a
231 // lifecycle event so the durable record stays
232 // in sync.
233 sink.Transcript(trimmed)
234 _ = log.Append(NewEvent("stdout", map[string]any{"text": trimmed}))
235 }
236 }
237 next:
238 if err != nil {
239 return
240 }
241 }
242 }()
243
244 // stderr: lifecycle/error chatter from the child. Every line
245 // is mirrored as a stderr event in the durable log AND surfaced
246 // in the transcript so users can diagnose a failing agent
247 // without leaving the dashboard.
248 go func() {
249 defer func() { done <- struct{}{} }()
250 br := bufio.NewReader(stderr)
251 for {
252 line, err := br.ReadString('\n')
253 if line != "" {
254 txt := strings.TrimRight(line, "\r\n")
255 sink.Transcript("stderr: " + txt)
256 _ = log.Append(NewEvent("stderr", map[string]any{"text": txt}))
257 }
258 if err != nil {
259 return
260 }
261 }
262 }()
263
264 <-done
265 <-done
266
267 err = cmd.Wait()
268 exit := 0
269 if ee, ok := err.(*exec.ExitError); ok {
270 exit = ee.ExitCode()
271 }
272 if err != nil && ctx.Err() != nil {
273 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"}))
274 return ctx.Err()
275 }
276 if err != nil {
277 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))
278 return err
279 }
280 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}))
281 sink.Activity("done")
282 return nil
283}
284
285// parseEventLine attempts to decode one JSONL line as an Event.
286// Returns ok=false for non-JSON or JSON without a "type" field.
287func parseEventLine(line string) (Event, bool) {
288 if len(line) == 0 || line[0] != '{' {
289 return Event{}, false
290 }
291 var ev Event
292 if err := json.Unmarshal([]byte(line), &ev); err != nil {
293 return Event{}, false
294 }
295 if ev.Type == "" {
296 return Event{}, false
297 }
298 if ev.Time.IsZero() {
299 ev.Time = time.Now()
300 }
301 return ev, true
302}
303
304// applyEventToSink translates an Event into Sink updates. Only a
305// few event types are interpreted; the rest still land in the
306// durable log via the caller.
307func applyEventToSink(ev Event, sink Sink) {
308 switch ev.Type {
309 case "assistant_message":
310 if c, ok := ev.Data["content"].([]any); ok {
311 for _, blk := range c {
312 m, _ := blk.(map[string]any)
313 if t, _ := m["type"].(string); t == "text" {
314 if txt, _ := m["text"].(string); txt != "" {
315 sink.Transcript(txt)
316 }
317 }
318 }
319 }
320 sink.Activity("idle")
321 case "user_message":
322 if c, ok := ev.Data["content"].([]any); ok {
323 for _, blk := range c {
324 m, _ := blk.(map[string]any)
325 if t, _ := m["type"].(string); t == "text" {
326 if txt, _ := m["text"].(string); txt != "" {
327 sink.Transcript("user: " + txt)
328 }
329 }
330 }
331 }
332 case "turn_start":
333 sink.Activity("thinking")
334 case "tool_call":
335 if name, _ := ev.Data["name"].(string); name != "" {
336 sink.Activity("tool: " + truncate(name, 60))
337 }
338 case "tool_result":
339 sink.Activity("idle")
340 case "turn_end":
341 sink.Activity("idle")
342 case "agent_ready":
343 sink.Activity("idle")
344 case "agent_stopped":
345 // terminal status is decided by Swarm.run from the runner's
346 // return value, not from this event. Don't overwrite the
347 // activity here.
348 case "error":
349 if msg, _ := ev.Data["message"].(string); msg != "" {
350 sink.Transcript("error: " + msg)
351 }
352 }
353}
354
355// RunnerFunc adapts a plain function into a Runner. Useful for tests
356// and for callers who don't need their own type.
357type RunnerFunc func(ctx context.Context, sink Sink) error
358
359func (f RunnerFunc) Run(ctx context.Context, sink Sink) error { return f(ctx, sink) }
360
361// streamLines kept around for any caller still using it directly.
362//
363// Deprecated: the runner now parses JSONL from stdout via
364// parseEventLine; this helper is unused inside the package but
365// remains exported via internal use by tests in the runner_test
366// suite that pre-date the daemon switch.
367func streamLines(r io.Reader, fn func(string)) {
368 br := bufio.NewReader(r)
369 for {
370 line, err := br.ReadString('\n')
371 if line != "" {
372 fn(strings.TrimRight(line, "\r\n"))
373 }
374 if err != nil {
375 return
376 }
377 }
378}