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