1package swarm
2
3// On-disk persistence for swarm agents.
4//
5// Every Spawn writes a meta.json next to the agent's events.jsonl and
6// session.json. The file captures the immutable identity bits (id,
7// task, branch, dir) plus the paths the runner needs to resume the
8// agent later. On a fresh zot launch, Swarm.Reload() walks
9// <root>/agents/*/meta.json and re-registers every agent it finds in
10// StatusDetached so the user can see, view, resume, or remove them
11// from the dashboard.
12//
13// We don't try to keep meta.json in sync with mutable state (status,
14// activity, transcript). Those live in the events log (durable) and
15// in-memory Agent fields (rebuilt by Reload from the log tail).
16// Keeping meta.json immutable means we never have to worry about
17// concurrent writers stomping on each other and the file matters
18// only on the spawn/reload boundary.
19
20import (
21 "context"
22 "encoding/json"
23 "errors"
24 "fmt"
25 "os"
26 "path/filepath"
27 "sort"
28 "strings"
29 "time"
30)
31
32// agentMeta is the durable identity record for one agent. Only fields
33// the supervisor needs to rebuild an Agent after a restart live here.
34// Adding a field is backwards-compatible (older meta.json files just
35// leave it zero); removing or renaming one is not.
36//
37// Historical fields like `branch` and `isolated` are silently dropped
38// by encoding/json's permissive decoder when an older meta.json is
39// loaded; we don't need to keep them in the struct.
40type agentMeta struct {
41 ID string `json:"id"`
42 Task string `json:"task"`
43 Dir string `json:"dir"`
44 Started time.Time `json:"started"`
45 Model string `json:"model,omitempty"`
46 Provider string `json:"provider,omitempty"`
47 InboxPath string `json:"inbox_path"`
48 EventLogPath string `json:"event_log_path"`
49 SessionPath string `json:"session_path"`
50
51 // SessionID, when non-empty, scopes the agent to a particular
52 // host zot session: the dashboard only shows agents whose
53 // SessionID matches the active session. Older meta.json files
54 // (and agents spawned outside of any session, e.g. by tests or
55 // scripted callers that didn't call SetActiveSession) have an
56 // empty SessionID and are visible from every session as a
57 // backward-compat fallback.
58 SessionID string `json:"session_id,omitempty"`
59}
60
61func metaPath(stateDir string) string { return filepath.Join(stateDir, "meta.json") }
62
63// writeAgentMeta serialises a's identity into stateDir/meta.json. The
64// write is atomic (tmp + rename) so a crash mid-write can't leave a
65// half-parsable file that fails Reload.
66func writeAgentMeta(stateDir string, a *Agent) error {
67 m := agentMeta{
68 ID: a.ID,
69 Task: a.Task,
70 Dir: a.Dir,
71 Started: a.Started,
72 Model: a.Model,
73 Provider: a.Provider,
74 InboxPath: a.InboxPath,
75 EventLogPath: a.EventLogPath,
76 SessionPath: a.SessionPath,
77 SessionID: a.SessionID,
78 }
79 b, err := json.MarshalIndent(m, "", " ")
80 if err != nil {
81 return fmt.Errorf("swarm meta marshal: %w", err)
82 }
83 if err := os.MkdirAll(stateDir, 0o755); err != nil {
84 return fmt.Errorf("swarm meta dir: %w", err)
85 }
86 final := metaPath(stateDir)
87 tmp := final + ".tmp"
88 if err := os.WriteFile(tmp, append(b, '\n'), 0o644); err != nil {
89 return fmt.Errorf("swarm meta write: %w", err)
90 }
91 if err := os.Rename(tmp, final); err != nil {
92 _ = os.Remove(tmp)
93 return fmt.Errorf("swarm meta rename: %w", err)
94 }
95 return nil
96}
97
98// readAgentMeta loads one meta.json. Returns os.ErrNotExist (wrapped)
99// when the file is missing so callers can distinguish "no such agent"
100// from "corrupt metadata".
101func readAgentMeta(stateDir string) (agentMeta, error) {
102 var m agentMeta
103 b, err := os.ReadFile(metaPath(stateDir))
104 if err != nil {
105 return m, err
106 }
107 if err := json.Unmarshal(b, &m); err != nil {
108 return m, fmt.Errorf("swarm meta parse %s: %w", stateDir, err)
109 }
110 if m.ID == "" {
111 return m, fmt.Errorf("swarm meta %s: missing id", stateDir)
112 }
113 return m, nil
114}
115
116// Reload scans <root>/agents/*/meta.json and re-registers every
117// previously-spawned agent as a StatusDetached entry. Agents already
118// present in memory are left alone (Reload is idempotent and safe to
119// call after Spawn, though in practice the cli invokes it exactly
120// once just after New()).
121//
122// The reloaded agents have no live Runner; the user can:
123// - view their transcript (the dashboard reads from EventLogPath),
124// - resume them via Swarm.Resume (starts a fresh subprocess on the
125// same worktree / session / inbox path),
126// - remove them (worktree + meta + events log gone).
127//
128// Reload returns the number of agents loaded plus any per-directory
129// error encountered. Malformed entries are skipped rather than
130// failing the whole reload — one bad meta.json shouldn't hide the
131// rest of the swarm.
132func (f *Swarm) Reload() (loaded int, errs []error) {
133 agentsDir := filepath.Join(f.cfg.Root, "agents")
134 entries, err := os.ReadDir(agentsDir)
135 if err != nil {
136 if errors.Is(err, os.ErrNotExist) {
137 return 0, nil
138 }
139 return 0, []error{fmt.Errorf("swarm reload: %w", err)}
140 }
141
142 // Sort by directory name so the load order is stable across runs.
143 // agentStateDir uses the id verbatim so name-sort == id-sort,
144 // which mirrors the creation order well enough for the dashboard
145 // (we also sort by Started in SnapshotAll, but having a stable
146 // f.order helps tests).
147 names := make([]string, 0, len(entries))
148 for _, e := range entries {
149 if e.IsDir() {
150 names = append(names, e.Name())
151 }
152 }
153 sort.Strings(names)
154
155 for _, name := range names {
156 stateDir := filepath.Join(agentsDir, name)
157 m, err := readAgentMeta(stateDir)
158 if err != nil {
159 if errors.Is(err, os.ErrNotExist) {
160 // Bare directory with no meta.json — probably a
161 // leftover from a Spawn that failed before
162 // writeAgentMeta. Ignore silently.
163 continue
164 }
165 errs = append(errs, err)
166 continue
167 }
168
169 f.mu.Lock()
170 if _, exists := f.agents[m.ID]; exists {
171 f.mu.Unlock()
172 continue
173 }
174 a := f.buildDetachedAgent(m)
175 f.agents[m.ID] = a
176 f.order = append(f.order, m.ID)
177 f.mu.Unlock()
178 loaded++
179 }
180 return loaded, errs
181}
182
183// detachedReplayBytes bounds startup replay; full history is loaded
184// when the dashboard is opened or the agent is resumed.
185const detachedReplayBytes = 4 << 20
186
187// buildDetachedAgent constructs an Agent from a meta.json with no
188// running Runner. The agent's transcript is populated from the tail
189// of its event log so the dashboard immediately shows recent output;
190// activity is inferred from the last lifecycle event.
191//
192// The returned Agent has a closed `done` channel because Wait should
193// return instantly: there is nothing to wait for.
194func (f *Swarm) buildDetachedAgent(m agentMeta) *Agent {
195 // Older meta.json files may still record a per-agent worktree
196 // path under Dir. They predate the decision to run every agent
197 // in the host's repo and shouldn't continue editing that stale
198 // checkout, which most likely no longer matches HEAD. Coerce
199 // the dir back to the live RepoRoot so resume picks up where
200 // the host is now.
201 dir := m.Dir
202 if f.cfg.RepoRoot != "" {
203 dir = f.cfg.RepoRoot
204 }
205 a := &Agent{
206 ID: m.ID,
207 Task: m.Task,
208 Dir: dir,
209 Started: m.Started,
210 Model: m.Model,
211 Provider: m.Provider,
212 InboxPath: m.InboxPath,
213 EventLogPath: m.EventLogPath,
214 SessionPath: m.SessionPath,
215 SessionID: m.SessionID,
216 inbox: NewInbox(m.InboxPath),
217 status: StatusDetached,
218 activity: "detached",
219 done: make(chan struct{}),
220 }
221 // Wait() must not block for detached agents; close the channel
222 // immediately. Callers Resuming the agent will replace done with
223 // a fresh channel before starting the new runner.
224 close(a.done)
225
226 // Recover transcript + activity hints from the event log. Best
227 // effort: a missing or unreadable log just leaves the agent
228 // detached with an empty transcript.
229 if a.EventLogPath != "" {
230 if evs, err := ReadEventLogTail(a.EventLogPath, detachedReplayBytes); err == nil {
231 replayEventsIntoAgent(a, evs)
232 }
233 if fi, err := os.Stat(a.EventLogPath); err == nil && fi.Size() > detachedReplayBytes {
234 a.needsFullReplay = true
235 }
236 }
237 return a
238}
239
240// loadFullTranscript backfills a detached agent on demand. Keep the
241// status inferred from the newest lifecycle event in the startup tail.
242// The agent's mutex serializes concurrent dashboard refreshes and resume.
243func (a *Agent) loadFullTranscript() {
244 a.mu.Lock()
245 defer a.mu.Unlock()
246 if !a.needsFullReplay {
247 return
248 }
249 evs, err := ReadEventLog(a.EventLogPath)
250 if err != nil {
251 return
252 }
253 full := &Agent{status: StatusDetached, activity: "detached"}
254 replayEventsIntoAgent(full, evs)
255 a.transcript = full.transcript
256 a.lastAssistant = full.lastAssistant
257 a.needsFullReplay = false
258}
259
260// replayEventsIntoAgent re-derives an agent's transcript and last
261// known status hint from its event log. Mirrors applyEventToSink in
262// runner.go but writes directly to the Agent fields (no Sink because
263// the agent isn't being driven by a runner yet).
264//
265// Status precedence: explicit lifecycle events (agent_stopped) win
266// over inferred ones (assistant_message → idle). If the log contains
267// no terminator we keep status=StatusDetached so the user can resume.
268func replayEventsIntoAgent(a *Agent, evs []Event) {
269 terminal := false
270 for _, ev := range evs {
271 switch ev.Type {
272 case "assistant_message", "user_message":
273 var text []string
274 if c, ok := ev.Data["content"].([]any); ok {
275 for _, blk := range c {
276 m, _ := blk.(map[string]any)
277 if t, _ := m["type"].(string); t == "text" {
278 if txt, _ := m["text"].(string); txt != "" {
279 text = append(text, txt)
280 }
281 }
282 }
283 }
284 message := strings.Join(text, "\n")
285 if ev.Type == "assistant_message" {
286 a.appendAssistantMessage(message)
287 } else {
288 a.appendUserMessage(message)
289 }
290 case "stdout":
291 if txt, _ := ev.Data["text"].(string); txt != "" {
292 a.appendTranscript(txt)
293 }
294 case "stderr":
295 if txt, _ := ev.Data["text"].(string); txt != "" {
296 a.appendTranscript("stderr: " + txt)
297 }
298 case "error":
299 if msg, _ := ev.Data["message"].(string); msg != "" {
300 a.appendTranscript("error: " + msg)
301 }
302 case "agent_stopped":
303 terminal = true
304 reason, _ := ev.Data["reason"].(string)
305 a.mu.Lock()
306 switch reason {
307 case "cancelled":
308 a.status = StatusKilled
309 a.activity = "cancelled (offline)"
310 case "shutdown":
311 a.status = StatusDone
312 a.activity = "shutdown (offline)"
313 case "exit":
314 if code, ok := ev.Data["code"].(float64); ok && code != 0 {
315 a.status = StatusFailed
316 a.activity = fmt.Sprintf("exit %d (offline)", int(code))
317 } else {
318 a.status = StatusDone
319 a.activity = "done (offline)"
320 }
321 default:
322 a.status = StatusDone
323 a.activity = "stopped (offline)"
324 }
325 a.mu.Unlock()
326 }
327 }
328 if !terminal {
329 // Non-terminal log means the previous parent died mid-run.
330 // Leave status=StatusDetached but record a hint so the
331 // dashboard shows something useful.
332 a.mu.Lock()
333 if a.activity == "detached" && len(a.transcript) > 0 {
334 a.activity = "detached (resume to continue)"
335 }
336 a.mu.Unlock()
337 }
338}
339
340// Resume re-attaches a Runner to a previously-spawned agent. Durable
341// session and event files are kept, while the transient inbox path is
342// recalculated for the current runtime environment. Use this to
343// continue a swarm session across zot restarts:
344//
345// swarmMgr.Reload()
346// a, err := swarmMgr.Resume(ctx, "alpha-12345")
347// swarmMgr.SendUserTurn(a.ID, "where were we?")
348//
349// The agent must be in a non-running state (Detached, Done, Failed,
350// Killed). Resuming a still-running agent returns an error so two
351// runners don't race for the same session.
352func (f *Swarm) Resume(ctx context.Context, id string) (*Agent, error) {
353 existing := f.Get(id)
354 if existing == nil {
355 return nil, fmt.Errorf("swarm: no such agent %q", id)
356 }
357 existing.mu.Lock()
358 st := existing.status
359 existing.mu.Unlock()
360 if st == StatusRunning || st == StatusPending {
361 return nil, fmt.Errorf("swarm: agent %s is still %s; stop it first", existing.ID, st)
362 }
363 existing.loadFullTranscript()
364
365 // Rebuild from the meta record so we don't carry stale runner
366 // state from a previous incarnation. The inbox is transient and
367 // may have been persisted under an incompatible filesystem by an
368 // older zot version, so always select it again on resume.
369 inboxPath, err := inboxSocketPath(f.cfg.Root, existing.ID)
370 if err != nil {
371 return nil, fmt.Errorf("swarm inbox path: %w", err)
372 }
373 m := agentMeta{
374 ID: existing.ID, Task: existing.Task,
375 Dir: existing.Dir, Started: existing.Started,
376 Model: existing.Model, Provider: existing.Provider,
377 InboxPath: inboxPath, EventLogPath: existing.EventLogPath,
378 SessionPath: existing.SessionPath,
379 SessionID: existing.SessionID,
380 }
381
382 a := &Agent{
383 ID: m.ID,
384 Task: m.Task,
385 Dir: m.Dir,
386 Started: m.Started,
387 Model: m.Model,
388 Provider: m.Provider,
389 InboxPath: m.InboxPath,
390 EventLogPath: m.EventLogPath,
391 SessionPath: m.SessionPath,
392 Resuming: true,
393 SessionID: m.SessionID,
394 inbox: NewInbox(m.InboxPath),
395 status: StatusPending,
396 activity: "resuming",
397 done: make(chan struct{}),
398 }
399 // Carry the previous transcript forward so the dashboard doesn't
400 // flash empty between resume and the first new event.
401 prev := existing.Transcript()
402 if len(prev) > 0 {
403 a.appendTranscript(strings.Join(prev, "\n"))
404 }
405 a.ctx, a.cancel = context.WithCancel(ctx)
406 a.runner = f.cfg.NewRunner(a)
407
408 f.mu.Lock()
409 f.agents[a.ID] = a
410 // Keep the agent's slot in f.order; replacing in-place avoids
411 // reshuffling the dashboard's row ordering on resume.
412 found := false
413 for _, k := range f.order {
414 if k == a.ID {
415 found = true
416 break
417 }
418 }
419 if !found {
420 f.order = append(f.order, a.ID)
421 }
422 f.mu.Unlock()
423
424 // Refresh the meta.json so any new path defaults (e.g. socket
425 // path moved into /tmp because the root got renamed) get
426 // persisted. Best-effort; resume still works if the disk is
427 // read-only because everything the runner needs is in-memory.
428 _ = writeAgentMeta(f.agentStateDir(a.ID), a)
429
430 go f.run(a)
431 return a, nil
432}