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