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