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 prompt-level task completions up to any
213 // subscriber on the supervised Agent. The child
214 // also forwards provider/tool-loop turn_end
215 // events (for example stop=tool_use); those do
216 // not contain step and must not be treated as
217 // swarm task completion.
218 notifyPromptTurnEnd(r.agent, ev)
219 } else {
220 // Non-JSON output. Keep it as transcript so an
221 // accidental fmt.Println in the child still
222 // shows up in the dashboard, and log a
223 // lifecycle event so the durable record stays
224 // in sync.
225 sink.Transcript(trimmed)
226 _ = log.Append(NewEvent("stdout", map[string]any{"text": trimmed}))
227 }
228 }
229 next:
230 if err != nil {
231 return
232 }
233 }
234 }()
235
236 // stderr: lifecycle/error chatter from the child. Every line
237 // is mirrored as a stderr event in the durable log AND surfaced
238 // in the transcript so users can diagnose a failing agent
239 // without leaving the dashboard.
240 go func() {
241 defer func() { done <- struct{}{} }()
242 br := bufio.NewReader(stderr)
243 for {
244 line, err := br.ReadString('\n')
245 if line != "" {
246 txt := strings.TrimRight(line, "\r\n")
247 sink.Transcript("stderr: " + txt)
248 _ = log.Append(NewEvent("stderr", map[string]any{"text": txt}))
249 }
250 if err != nil {
251 return
252 }
253 }
254 }()
255
256 <-done
257 <-done
258
259 err = cmd.Wait()
260 exit := 0
261 if ee, ok := err.(*exec.ExitError); ok {
262 exit = ee.ExitCode()
263 }
264 if err != nil && ctx.Err() != nil {
265 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "cancelled"}))
266 return ctx.Err()
267 }
268 if err != nil {
269 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": exit, "error": err.Error()}))
270 return err
271 }
272 _ = log.Append(NewEvent("agent_stopped", map[string]any{"reason": "exit", "code": 0}))
273 sink.Activity("done")
274 return nil
275}
276
277// parseEventLine attempts to decode one JSONL line as an Event.
278// Returns ok=false for non-JSON or JSON without a "type" field.
279func parseEventLine(line string) (Event, bool) {
280 if len(line) == 0 || line[0] != '{' {
281 return Event{}, false
282 }
283 var ev Event
284 if err := json.Unmarshal([]byte(line), &ev); err != nil {
285 return Event{}, false
286 }
287 if ev.Type == "" {
288 return Event{}, false
289 }
290 if ev.Time.IsZero() {
291 ev.Time = time.Now()
292 }
293 return ev, true
294}
295
296// notifyPromptTurnEnd calls Agent.OnTurnEnd only for the swarm
297// daemon's prompt-level completion event. Provider/tool-loop
298// turn_end events (such as stop=tool_use) do not include step and
299// are not terminal for the delegated task.
300func notifyPromptTurnEnd(a *Agent, ev Event) {
301 if a == nil || ev.Type != "turn_end" {
302 return
303 }
304 step, ok := ev.Data["step"].(float64)
305 if !ok {
306 return
307 }
308
309 a.mu.Lock()
310 fn := a.OnTurnEnd
311 a.mu.Unlock()
312 if fn != nil {
313 errMsg, _ := ev.Data["error"].(string)
314 go fn(int(step), errMsg)
315 }
316}
317
318// applyEventToSink translates an Event into Sink updates. Only a
319// few event types are interpreted; the rest still land in the
320// durable log via the caller.
321func applyEventToSink(ev Event, sink Sink) {
322 switch ev.Type {
323 case "assistant_message":
324 if c, ok := ev.Data["content"].([]any); ok {
325 for _, blk := range c {
326 m, _ := blk.(map[string]any)
327 if t, _ := m["type"].(string); t == "text" {
328 if txt, _ := m["text"].(string); txt != "" {
329 sink.Transcript(txt)
330 }
331 }
332 }
333 }
334 sink.Activity("idle")
335 case "user_message":
336 if c, ok := ev.Data["content"].([]any); ok {
337 for _, blk := range c {
338 m, _ := blk.(map[string]any)
339 if t, _ := m["type"].(string); t == "text" {
340 if txt, _ := m["text"].(string); txt != "" {
341 sink.Transcript("user: " + txt)
342 }
343 }
344 }
345 }
346 case "turn_start":
347 sink.Activity("thinking")
348 case "tool_call":
349 if name, _ := ev.Data["name"].(string); name != "" {
350 sink.Activity("tool: " + truncate(name, 60))
351 }
352 case "tool_result":
353 sink.Activity("idle")
354 case "turn_end":
355 sink.Activity("idle")
356 case "agent_ready":
357 sink.Activity("idle")
358 case "agent_stopped":
359 // terminal status is decided by Swarm.run from the runner's
360 // return value, not from this event. Don't overwrite the
361 // activity here.
362 case "error":
363 if msg, _ := ev.Data["message"].(string); msg != "" {
364 sink.Transcript("error: " + msg)
365 }
366 }
367}
368
369// RunnerFunc adapts a plain function into a Runner. Useful for tests
370// and for callers who don't need their own type.
371type RunnerFunc func(ctx context.Context, sink Sink) error
372
373func (f RunnerFunc) Run(ctx context.Context, sink Sink) error { return f(ctx, sink) }
374
375// streamLines kept around for any caller still using it directly.
376//
377// Deprecated: the runner now parses JSONL from stdout via
378// parseEventLine; this helper is unused inside the package but
379// remains exported via internal use by tests in the runner_test
380// suite that pre-date the daemon switch.
381func streamLines(r io.Reader, fn func(string)) {
382 br := bufio.NewReader(r)
383 for {
384 line, err := br.ReadString('\n')
385 if line != "" {
386 fn(strings.TrimRight(line, "\r\n"))
387 }
388 if err != nil {
389 return
390 }
391 }
392}