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