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// buildDetachedAgent constructs an Agent from a meta.json with no
184// running Runner. The agent's transcript is populated from the tail
185// of its event log so the dashboard immediately shows recent output;
186// activity is inferred from the last lifecycle event.
187//
188// The returned Agent has a closed `done` channel because Wait should
189// return instantly: there is nothing to wait for.
190func (f *Swarm) buildDetachedAgent(m agentMeta) *Agent {
191 // Older meta.json files may still record a per-agent worktree
192 // path under Dir. They predate the decision to run every agent
193 // in the host's repo and shouldn't continue editing that stale
194 // checkout, which most likely no longer matches HEAD. Coerce
195 // the dir back to the live RepoRoot so resume picks up where
196 // the host is now.
197 dir := m.Dir
198 if f.cfg.RepoRoot != "" {
199 dir = f.cfg.RepoRoot
200 }
201 a := &Agent{
202 ID: m.ID,
203 Task: m.Task,
204 Dir: dir,
205 Started: m.Started,
206 Model: m.Model,
207 Provider: m.Provider,
208 InboxPath: m.InboxPath,
209 EventLogPath: m.EventLogPath,
210 SessionPath: m.SessionPath,
211 SessionID: m.SessionID,
212 inbox: NewInbox(m.InboxPath),
213 status: StatusDetached,
214 activity: "detached",
215 done: make(chan struct{}),
216 }
217 // Wait() must not block for detached agents; close the channel
218 // immediately. Callers Resuming the agent will replace done with
219 // a fresh channel before starting the new runner.
220 close(a.done)
221
222 // Recover transcript + activity hints from the event log. Best
223 // effort: a missing or unreadable log just leaves the agent
224 // detached with an empty transcript.
225 if a.EventLogPath != "" {
226 if evs, err := ReadEventLog(a.EventLogPath); err == nil {
227 replayEventsIntoAgent(a, evs)
228 }
229 }
230 return a
231}
232
233// replayEventsIntoAgent re-derives an agent's transcript and last
234// known status hint from its event log. Mirrors applyEventToSink in
235// runner.go but writes directly to the Agent fields (no Sink because
236// the agent isn't being driven by a runner yet).
237//
238// Status precedence: explicit lifecycle events (agent_stopped) win
239// over inferred ones (assistant_message → idle). If the log contains
240// no terminator we keep status=StatusDetached so the user can resume.
241func replayEventsIntoAgent(a *Agent, evs []Event) {
242 terminal := false
243 for _, ev := range evs {
244 switch ev.Type {
245 case "assistant_message":
246 if c, ok := ev.Data["content"].([]any); ok {
247 for _, blk := range c {
248 m, _ := blk.(map[string]any)
249 if t, _ := m["type"].(string); t == "text" {
250 if txt, _ := m["text"].(string); txt != "" {
251 a.appendTranscript(txt)
252 }
253 }
254 }
255 }
256 case "user_message":
257 if c, ok := ev.Data["content"].([]any); ok {
258 for _, blk := range c {
259 m, _ := blk.(map[string]any)
260 if t, _ := m["type"].(string); t == "text" {
261 if txt, _ := m["text"].(string); txt != "" {
262 a.appendTranscript("user: " + txt)
263 }
264 }
265 }
266 }
267 case "stdout":
268 if txt, _ := ev.Data["text"].(string); txt != "" {
269 a.appendTranscript(txt)
270 }
271 case "stderr":
272 if txt, _ := ev.Data["text"].(string); txt != "" {
273 a.appendTranscript("stderr: " + txt)
274 }
275 case "error":
276 if msg, _ := ev.Data["message"].(string); msg != "" {
277 a.appendTranscript("error: " + msg)
278 }
279 case "agent_stopped":
280 terminal = true
281 reason, _ := ev.Data["reason"].(string)
282 a.mu.Lock()
283 switch reason {
284 case "cancelled":
285 a.status = StatusKilled
286 a.activity = "cancelled (offline)"
287 case "shutdown":
288 a.status = StatusDone
289 a.activity = "shutdown (offline)"
290 case "exit":
291 if code, ok := ev.Data["code"].(float64); ok && code != 0 {
292 a.status = StatusFailed
293 a.activity = fmt.Sprintf("exit %d (offline)", int(code))
294 } else {
295 a.status = StatusDone
296 a.activity = "done (offline)"
297 }
298 default:
299 a.status = StatusDone
300 a.activity = "stopped (offline)"
301 }
302 a.mu.Unlock()
303 }
304 }
305 if !terminal {
306 // Non-terminal log means the previous parent died mid-run.
307 // Leave status=StatusDetached but record a hint so the
308 // dashboard shows something useful.
309 a.mu.Lock()
310 if a.activity == "detached" && len(a.transcript) > 0 {
311 a.activity = "detached (resume to continue)"
312 }
313 a.mu.Unlock()
314 }
315}
316
317// Resume re-attaches a Runner to a previously-spawned agent. The
318// existing worktree, session file, branch, and inbox path are kept;
319// only the in-memory Agent and its runner are replaced. Use this to
320// continue a swarm session across zot restarts:
321//
322// swarmMgr.Reload()
323// a, err := swarmMgr.Resume(ctx, "alpha-12345")
324// swarmMgr.SendUserTurn(a.ID, "where were we?")
325//
326// The agent must be in a non-running state (Detached, Done, Failed,
327// Killed). Resuming a still-running agent returns an error so two
328// runners don't race for the same session.
329func (f *Swarm) Resume(ctx context.Context, id string) (*Agent, error) {
330 existing := f.Get(id)
331 if existing == nil {
332 return nil, fmt.Errorf("swarm: no such agent %q", id)
333 }
334 existing.mu.Lock()
335 st := existing.status
336 existing.mu.Unlock()
337 if st == StatusRunning || st == StatusPending {
338 return nil, fmt.Errorf("swarm: agent %s is still %s; stop it first", existing.ID, st)
339 }
340
341 // Rebuild from the meta record so we don't carry stale runner
342 // state from a previous incarnation. We re-read meta.json rather
343 // than reusing the live struct's fields so callers that mutated
344 // (e.g. tests that hand-built an Agent) don't accidentally route
345 // the new runner at the wrong paths.
346 m := agentMeta{
347 ID: existing.ID, Task: existing.Task,
348 Dir: existing.Dir, Started: existing.Started,
349 Model: existing.Model, Provider: existing.Provider,
350 InboxPath: existing.InboxPath, EventLogPath: existing.EventLogPath,
351 SessionPath: existing.SessionPath,
352 }
353
354 a := &Agent{
355 ID: m.ID,
356 Task: m.Task,
357 Dir: m.Dir,
358 Started: m.Started,
359 Model: m.Model,
360 Provider: m.Provider,
361 InboxPath: m.InboxPath,
362 EventLogPath: m.EventLogPath,
363 SessionPath: m.SessionPath,
364 Resuming: true,
365 inbox: NewInbox(m.InboxPath),
366 status: StatusPending,
367 activity: "resuming",
368 done: make(chan struct{}),
369 }
370 // Carry the previous transcript forward so the dashboard doesn't
371 // flash empty between resume and the first new event.
372 prev := existing.Transcript()
373 if len(prev) > 0 {
374 a.appendTranscript(strings.Join(prev, "\n"))
375 }
376 a.ctx, a.cancel = context.WithCancel(ctx)
377 a.runner = f.cfg.NewRunner(a)
378
379 f.mu.Lock()
380 f.agents[a.ID] = a
381 // Keep the agent's slot in f.order; replacing in-place avoids
382 // reshuffling the dashboard's row ordering on resume.
383 found := false
384 for _, k := range f.order {
385 if k == a.ID {
386 found = true
387 break
388 }
389 }
390 if !found {
391 f.order = append(f.order, a.ID)
392 }
393 f.mu.Unlock()
394
395 // Refresh the meta.json so any new path defaults (e.g. socket
396 // path moved into /tmp because the root got renamed) get
397 // persisted. Best-effort; resume still works if the disk is
398 // read-only because everything the runner needs is in-memory.
399 _ = writeAgentMeta(f.agentStateDir(a.ID), a)
400
401 go f.run(a)
402 return a, nil
403}