1package subagents
2
3// On-disk persistence for subagent agents.
4//
5// Every Spawn writes a meta.json next to the agent's events.jsonl and
6// session.json. The file captures identity, resumable lifecycle state, and
7// compact dashboard state. On a fresh zut launch, Supervisor.Reload() walks
8// <root>/agents/*/meta.json and re-registers every agent without replaying
9// its full history. The event log remains the durable transcript source and
10// is hydrated only when a caller requests that agent's transcript.
11//
12// Metadata writes are atomic (tmp + rename). The supervisor is the sole
13// writer, so its lifecycle snapshots cannot race with another host process.
14
15import (
16 "context"
17 "encoding/json"
18 "errors"
19 "fmt"
20 "os"
21 "path/filepath"
22 "sort"
23 "strings"
24 "time"
25)
26
27// agentMeta is the durable identity record for one agent. Only fields
28// the supervisor needs to rebuild an Agent after a restart live here.
29// Adding a field leaves it zero when absent; removing or renaming a field
30// changes the persisted contract.
31//
32// Historical fields like `branch` and `isolated` are silently dropped
33// by encoding/json's permissive decoder when an older meta.json is
34// loaded; we don't need to keep them in the struct.
35type agentMeta struct {
36 ID string `json:"id"`
37 Task string `json:"task"`
38 OriginalTask string `json:"original_task,omitempty"`
39 Dir string `json:"dir"`
40 RepositoryRoot string `json:"repository_root,omitempty"`
41 Started time.Time `json:"started"`
42 ParentID string `json:"parent_id,omitempty"`
43 BatchID string `json:"batch_id,omitempty"`
44 RootSessionID string `json:"root_session_id,omitempty"`
45 Model string `json:"model,omitempty"`
46 Provider string `json:"provider,omitempty"`
47 BaseURL string `json:"base_url,omitempty"`
48 InsecureTLS bool `json:"insecure_tls,omitempty"`
49 Reasoning string `json:"reasoning,omitempty"`
50 FastMode *bool `json:"fast_mode,omitempty"`
51 Subagent string `json:"subagent,omitempty"`
52 WorkspaceMode WorkspaceMode `json:"workspace_mode,omitempty"`
53 WorkspacePath string `json:"workspace_path,omitempty"`
54 WorkspaceBase string `json:"workspace_base,omitempty"`
55 WorkspaceCapture CaptureMode `json:"workspace_capture,omitempty"`
56 MaxTurns int `json:"max_turns,omitempty"`
57 Timeout time.Duration `json:"timeout,omitempty"`
58 Tools []string `json:"tools,omitempty"`
59 WebSearchPolicy WebSearchPolicy `json:"web_search_policy,omitempty"`
60 Status Status `json:"status,omitempty"`
61 Activity string `json:"activity,omitempty"`
62 Finished time.Time `json:"finished,omitempty"`
63 Error string `json:"error,omitempty"`
64 ProcessState ProcessState `json:"process_state,omitempty"`
65 TurnState TurnState `json:"turn_state,omitempty"`
66 CurrentTurnID string `json:"current_turn_id,omitempty"`
67 Attempt int `json:"attempt,omitempty"`
68 ProcessPID int `json:"process_pid,omitempty"`
69 UpdatedAt time.Time `json:"updated_at,omitempty"`
70 LastActivity time.Time `json:"last_activity,omitempty"`
71 ResultRef string `json:"result_ref,omitempty"`
72 PatchRef string `json:"patch_ref,omitempty"`
73 ChangedFiles []string `json:"changed_files,omitempty"`
74 InboxPath string `json:"inbox_path"`
75 EventLogPath string `json:"event_log_path"`
76 SessionPath string `json:"session_path"`
77 // ResumePrompt is a follow-up accepted by subagent_resume but not yet
78 // acknowledged by the child with a turn.started event. It is durable so a
79 // host exit while the worker is queued does not lose the manager signal.
80 ResumePrompt string `json:"resume_prompt,omitempty"`
81 ResumePromptAt time.Time `json:"resume_prompt_at,omitempty"`
82
83 // SessionID, when non-empty, scopes the agent to a particular
84 // host zut session: the dashboard only shows agents whose
85 // SessionID matches the active session. Older meta.json files
86 // (and agents spawned outside of any session, e.g. by tests or
87 // scripted callers that didn't call SetActiveSession) have an
88 // empty SessionID and are visible from every session as a
89 // unscoped-agent fallback.
90 SessionID string `json:"session_id,omitempty"`
91}
92
93func metaPath(stateDir string) string { return filepath.Join(stateDir, "meta.json") }
94
95// writeAgentMeta serialises a's identity into stateDir/meta.json. The
96// write is atomic (tmp + rename) so a crash mid-write can't leave a
97// half-parsable file that fails Reload.
98func writeAgentMeta(stateDir string, a *Agent) error {
99 fastMode := a.FastMode
100 webSearchPolicy := childWebSearchPolicy(a.WebSearchPolicy, a.Subagent, a.Tools)
101 a.mu.Lock()
102 status := a.status
103 activity := a.activity
104 finished := a.finished
105 resumePrompt := a.resumePromptText
106 resumePromptAt := a.resumePromptAt
107 errText := ""
108 if a.lastErr != nil {
109 errText = a.lastErr.Error()
110 }
111 a.mu.Unlock()
112 a.lifecycleMu.Lock()
113 processState := a.processState
114 turnState := a.turnState
115 currentTurnID := a.currentTurnID
116 attempt := a.Attempt
117 processPID := a.ProcessPID
118 updatedAt := a.updatedAt
119 lastActivity := a.lastActivity
120 resultRef := a.resultRef
121 patchRef := a.patchRef
122 changedFiles := append([]string(nil), a.changedFiles...)
123 a.lifecycleMu.Unlock()
124 m := agentMeta{
125 ID: a.ID,
126 Task: a.Task,
127 OriginalTask: a.OriginalTask,
128 Dir: a.Dir,
129 RepositoryRoot: a.RepositoryRoot,
130 Started: a.Started,
131 ParentID: a.ParentID,
132 BatchID: a.BatchID,
133 RootSessionID: a.RootSessionID,
134 Model: a.Model,
135 Provider: a.Provider,
136 BaseURL: a.BaseURL,
137 InsecureTLS: a.InsecureTLS,
138 Reasoning: a.Reasoning,
139 FastMode: &fastMode,
140 Subagent: a.Subagent,
141 WorkspaceMode: a.WorkspaceMode,
142 WorkspacePath: a.WorkspacePath,
143 WorkspaceBase: a.WorkspaceBase,
144 WorkspaceCapture: a.WorkspaceCapture,
145 MaxTurns: a.MaxTurns,
146 Timeout: a.Timeout,
147 Tools: append([]string(nil), a.Tools...),
148 WebSearchPolicy: webSearchPolicy,
149 Status: status,
150 Activity: activity,
151 Finished: finished,
152 Error: errText,
153 ProcessState: processState,
154 TurnState: turnState,
155 CurrentTurnID: currentTurnID,
156 Attempt: attempt,
157 ProcessPID: processPID,
158 UpdatedAt: updatedAt,
159 LastActivity: lastActivity,
160 ResultRef: resultRef,
161 PatchRef: patchRef,
162 ChangedFiles: changedFiles,
163 InboxPath: a.InboxPath,
164 EventLogPath: a.EventLogPath,
165 SessionPath: a.SessionPath,
166 ResumePrompt: resumePrompt,
167 ResumePromptAt: resumePromptAt,
168 SessionID: a.SessionID,
169 }
170 b, err := json.MarshalIndent(m, "", " ")
171 if err != nil {
172 return fmt.Errorf("subagent meta marshal: %w", err)
173 }
174 if err := os.MkdirAll(stateDir, 0o700); err != nil {
175 return fmt.Errorf("subagents meta dir: %w", err)
176 }
177 final := metaPath(stateDir)
178 tmp := final + ".tmp"
179 if err := os.WriteFile(tmp, append(b, '\n'), 0o600); err != nil {
180 return fmt.Errorf("subagents meta write: %w", err)
181 }
182 if err := os.Rename(tmp, final); err != nil {
183 _ = os.Remove(tmp)
184 return fmt.Errorf("subagent meta rename: %w", err)
185 }
186 return nil
187}
188
189// readAgentMeta loads one meta.json. Returns os.ErrNotExist (wrapped)
190// when the file is missing so callers can distinguish "no such agent"
191// from "corrupt metadata".
192func readAgentMeta(stateDir string) (agentMeta, error) {
193 var m agentMeta
194 b, err := os.ReadFile(metaPath(stateDir))
195 if err != nil {
196 return m, err
197 }
198 if err := json.Unmarshal(b, &m); err != nil {
199 return m, fmt.Errorf("subagent meta parse %s: %w", stateDir, err)
200 }
201 if m.ID == "" {
202 return m, fmt.Errorf("subagent meta %s: missing id", stateDir)
203 }
204 return m, nil
205}
206
207func safeAgentID(id string) bool {
208 return id != "" && id != "." && id != ".." && filepath.Base(id) == id && !strings.ContainsAny(id, `/\\`)
209}
210
211func (f *Supervisor) sanitizeReloadMeta(stateDir, root string, m *agentMeta) error {
212 if m == nil {
213 return errors.New("subagents reload: nil metadata")
214 }
215 if m.EventLogPath == "" {
216 m.EventLogPath = filepath.Join(stateDir, "events.jsonl")
217 }
218 if m.SessionPath == "" {
219 m.SessionPath = filepath.Join(stateDir, "session.json")
220 }
221 if filepath.Base(m.EventLogPath) != "events.jsonl" || !pathWithin(m.EventLogPath, stateDir) {
222 return fmt.Errorf("subagents reload %s: event log path escapes agent state", stateDir)
223 }
224 if filepath.Base(m.SessionPath) != "session.json" || !pathWithin(m.SessionPath, stateDir) {
225 return fmt.Errorf("subagents reload %s: session path escapes agent state", stateDir)
226 }
227 if m.InboxPath == "" || !validRuntimeInboxPath(m.InboxPath, m.ID) {
228 // Old metadata may point at a runtime directory from a previous
229 // environment. Recompute that transient path rather than ever
230 // dialing an arbitrary socket supplied by metadata.
231 expectedInbox, err := inboxSocketPath(root, m.ID)
232 if err != nil {
233 return fmt.Errorf("subagents reload %s: inbox path: %w", stateDir, err)
234 }
235 m.InboxPath = expectedInbox
236 }
237 return nil
238}
239
240func validRuntimeInboxPath(path, id string) bool {
241 if !filepath.IsAbs(path) || (filepath.Base(path) != id+".sock" && filepath.Base(path) != shortHash(id)+".sock") {
242 return false
243 }
244 dir := filepath.Dir(path)
245 if !strings.HasPrefix(filepath.Base(dir), "zut-subagents-") {
246 return false
247 }
248 bases := []string{os.TempDir(), "/tmp"}
249 if runtimeDir := os.Getenv("XDG_RUNTIME_DIR"); filepath.IsAbs(runtimeDir) {
250 bases = append(bases, runtimeDir)
251 }
252 for _, base := range bases {
253 if pathWithin(dir, filepath.Clean(base)) {
254 return true
255 }
256 }
257 return false
258}
259
260func firstNonEmpty(values ...string) string {
261 for _, value := range values {
262 if strings.TrimSpace(value) != "" {
263 return value
264 }
265 }
266 return ""
267}
268
269// Reload scans <root>/agents/*/meta.json and re-registers every
270// previously-spawned agent as a StatusDetached entry. Agents already
271// present in memory are left alone (Reload is idempotent and safe to
272// call after Spawn, though in practice the cli invokes it exactly
273// once just after New()).
274//
275// The reloaded agents have no live Runner; the user can:
276// - view their transcript (the dashboard reads from EventLogPath),
277// - resume them via Supervisor.Resume (starts a fresh subprocess on the
278// same worktree / session / inbox path),
279// - remove them (worktree + meta + events log gone).
280//
281// Reload returns the number of agents loaded plus any per-directory
282// error encountered. Malformed entries are skipped rather than
283// failing the whole reload — one bad meta.json shouldn't hide the
284// rest of the subagents.
285func (f *Supervisor) Reload() (loaded int, errs []error) {
286 root := filepath.Clean(f.cfg.Root)
287 if root == "." {
288 return 0, nil
289 }
290 loaded, errs = f.reloadRoot(root)
291 errs = append(errs, f.reloadBatches(root)...)
292 return loaded, errs
293}
294
295func (f *Supervisor) reloadRoot(root string) (loaded int, errs []error) {
296 agentsDir := filepath.Join(root, "agents")
297 entries, err := os.ReadDir(agentsDir)
298 if err != nil {
299 if errors.Is(err, os.ErrNotExist) {
300 return 0, nil
301 }
302 return 0, []error{fmt.Errorf("subagents reload %s: %w", root, err)}
303 }
304
305 // Sort by directory name so the load order is stable across runs.
306 names := make([]string, 0, len(entries))
307 for _, e := range entries {
308 if e.IsDir() {
309 names = append(names, e.Name())
310 }
311 }
312 sort.Strings(names)
313
314 for _, name := range names {
315 stateDir := filepath.Join(agentsDir, name)
316 m, err := readAgentMeta(stateDir)
317 if err != nil {
318 if errors.Is(err, os.ErrNotExist) {
319 continue
320 }
321 errs = append(errs, err)
322 continue
323 }
324 if filepath.Base(stateDir) != m.ID || !safeAgentID(m.ID) {
325 errs = append(errs, fmt.Errorf("subagents reload %s: unsafe or mismatched agent id", stateDir))
326 continue
327 }
328 if err := f.sanitizeReloadMeta(stateDir, root, &m); err != nil {
329 errs = append(errs, err)
330 continue
331 }
332
333 f.mu.Lock()
334 if _, exists := f.agents[m.ID]; exists {
335 f.mu.Unlock()
336 continue
337 }
338 a := f.buildDetachedAgent(m)
339 f.agents[m.ID] = a
340 f.order = append(f.order, m.ID)
341 f.mu.Unlock()
342 loaded++
343 }
344 return loaded, errs
345}
346
347// buildDetachedAgent constructs an Agent from compact metadata with no running
348// Runner. Its transcript stays unloaded until a caller requests it, while the
349// persisted lifecycle summary keeps the dashboard immediately useful.
350//
351// The returned Agent has a closed `done` channel because Wait should return
352// instantly: there is nothing to wait for.
353func (f *Supervisor) buildDetachedAgent(m agentMeta) *Agent {
354 // Metadata without a repository identity uses the current RepoRoot;
355 // records with one retain their persisted repository so a restart from
356 // another checkout cannot silently redirect the child.
357 dir := m.Dir
358 if m.WorkspaceMode != WorkspaceWorktree && m.RepositoryRoot == "" && f.cfg.RepoRoot != "" {
359 // Preserve the best-effort migration only for records without an
360 // explicit repository identity.
361 dir = f.cfg.RepoRoot
362 }
363 if dir == "" {
364 dir = firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot)
365 }
366 fastMode := f.cfg.FastMode
367 if m.FastMode != nil {
368 fastMode = *m.FastMode
369 }
370 turnState := m.TurnState
371 if turnState == "" {
372 turnState = TurnIdle
373 }
374 lastActivity := m.LastActivity
375 if lastActivity.IsZero() {
376 lastActivity = m.Started
377 }
378 persistedStatus := m.Status
379 status := StatusDetached
380 activity := strings.TrimSpace(m.Activity)
381 processState := m.ProcessState
382 legacyEventState := persistedStatus == ""
383 if legacyEventState {
384 activity = "detached"
385 processState = ProcessDetached
386 }
387 if processState == "" {
388 processState = ProcessDetached
389 }
390 if activity == "" {
391 switch persistedStatus {
392 case StatusDone:
393 activity = "done"
394 case StatusFailed:
395 activity = "failed"
396 case StatusKilled:
397 activity = "cancelled"
398 default:
399 activity = "detached"
400 }
401 }
402 if persistedStatus == StatusRunning || persistedStatus == StatusPending {
403 activity = "detached (resume to continue)"
404 processState = ProcessDetached
405 }
406 if inboxLive(m.InboxPath) {
407 processState = ProcessAlive
408 activity = "live (detached)"
409 }
410 var lastErr error
411 if m.Error != "" {
412 lastErr = errors.New(m.Error)
413 }
414 agentStateDir := f.agentStateDir(m.ID)
415 if m.EventLogPath != "" {
416 // EventLogPath is <state-dir>/events.jsonl; keep the reloaded
417 // agent's state scoped to its own directory. Using the parent
418 // agents/ directory here would make Remove delete sibling agents
419 // and would make resumed result/patch writes collide.
420 agentStateDir = filepath.Dir(m.EventLogPath)
421 }
422 a := &Agent{
423 ID: m.ID,
424 Task: m.Task,
425 OriginalTask: firstNonEmpty(m.OriginalTask, m.Task),
426 Dir: dir,
427 RepositoryRoot: firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot),
428 Started: m.Started,
429 ParentID: m.ParentID,
430 BatchID: m.BatchID,
431 RootSessionID: m.RootSessionID,
432 Model: m.Model,
433 Provider: m.Provider,
434 BaseURL: m.BaseURL,
435 InsecureTLS: m.InsecureTLS,
436 Reasoning: m.Reasoning,
437 FastMode: fastMode,
438 Subagent: m.Subagent,
439 WorkspaceMode: m.WorkspaceMode,
440 WorkspacePath: m.WorkspacePath,
441 WorkspaceBase: m.WorkspaceBase,
442 WorkspaceCapture: m.WorkspaceCapture,
443 MaxTurns: m.MaxTurns,
444 Timeout: m.Timeout,
445 Tools: append([]string(nil), m.Tools...),
446 WebSearchPolicy: childWebSearchPolicy(m.WebSearchPolicy, m.Subagent, m.Tools),
447 Attempt: m.Attempt,
448 ProcessPID: m.ProcessPID,
449 InboxPath: m.InboxPath,
450 EventLogPath: m.EventLogPath,
451 SessionPath: m.SessionPath,
452 resumePromptText: m.ResumePrompt,
453 resumePromptAt: m.ResumePromptAt,
454 SessionID: m.SessionID,
455 inbox: NewInbox(m.InboxPath),
456 status: status,
457 activity: activity,
458 finished: m.Finished,
459 lastErr: lastErr,
460 legacyEventState: legacyEventState,
461 processState: processState,
462 turnState: turnState,
463 currentTurnID: m.CurrentTurnID,
464 updatedAt: m.UpdatedAt,
465 lastActivity: lastActivity,
466 stateDir: agentStateDir,
467 resultRef: m.ResultRef,
468 patchRef: m.PatchRef,
469 changedFiles: append([]string(nil), m.ChangedFiles...),
470 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
471 maxOutputLines: f.cfg.Policy.MaxOutputLines,
472 done: make(chan struct{}),
473 turnResults: make(chan *TurnResult, 16),
474 }
475 if a.updatedAt.IsZero() {
476 a.updatedAt = lastActivity
477 }
478 a.closeDone()
479
480 // The append-only log may contain a terminal event from an earlier
481 // attempt. A live socket is stronger evidence for the current process;
482 // keep the reloaded entry detached/live so Stop and Remove cannot treat a
483 // resumed worker as already finished.
484 if inboxLive(a.InboxPath) {
485 a.mu.Lock()
486 a.status = StatusDetached
487 a.activity = "live (detached)"
488 a.mu.Unlock()
489 a.setProcessState(ProcessAlive)
490 }
491 resultDir := f.agentStateDir(a.ID)
492 if a.EventLogPath != "" {
493 resultDir = filepath.Dir(a.EventLogPath)
494 }
495 if result, err := readTurnResult(resultDir); err == nil && validateTurnResultAgent(result, a.ID) == nil {
496 a.setResult(result.Bounded(f.cfg.Policy.MaxOutputBytes, f.cfg.Policy.MaxOutputLines))
497 }
498 return a
499}
500
501// LoadTranscript hydrates one reloaded agent's durable event history. Reload
502// deliberately skips this work so startup stays bounded by metadata size;
503// callers request it only when opening the transcript or otherwise reading it.
504func (f *Supervisor) LoadTranscript(id string) error {
505 a := f.Get(id)
506 if a == nil {
507 return fmt.Errorf("subagent: no such agent %q", id)
508 }
509 return a.loadTranscript()
510}
511
512func (a *Agent) loadTranscript() error {
513 if a == nil {
514 return nil
515 }
516 a.transcriptMu.Lock()
517 defer a.transcriptMu.Unlock()
518
519 a.mu.Lock()
520 if a.transcriptLoaded {
521 a.mu.Unlock()
522 return nil
523 }
524 path := a.EventLogPath
525 legacyEventState := a.legacyEventState
526 a.mu.Unlock()
527 var summaryUpdatedAt, summaryLastActivity time.Time
528 if !legacyEventState {
529 a.lifecycleMu.Lock()
530 summaryUpdatedAt = a.updatedAt
531 summaryLastActivity = a.lastActivity
532 a.lifecycleMu.Unlock()
533 }
534 if path == "" {
535 a.mu.Lock()
536 a.transcriptLoaded = true
537 a.mu.Unlock()
538 return nil
539 }
540
541 events, err := ReadEventLog(path)
542 if err != nil {
543 return fmt.Errorf("read subagent transcript: %w", err)
544 }
545 if prompt, acceptedAt := a.ResumePromptInfo(); prompt != "" && resumePromptAcknowledged(events, acceptedAt) {
546 a.clearResumePrompt()
547 }
548 a.mu.Lock()
549 if a.transcriptLoaded {
550 a.mu.Unlock()
551 return nil
552 }
553 a.transcript = nil
554 a.lastAssistant = ""
555 a.outputBytes = 0
556 a.outputLines = 0
557 a.outputTruncated = false
558 a.mu.Unlock()
559
560 if legacyEventState {
561 // Metadata written before the dashboard summary existed has no
562 // terminal status to trust. Preserve its historical event-derived
563 // behavior once the user explicitly opens its transcript.
564 replayEventsIntoAgent(a, events)
565 } else {
566 replayTranscriptIntoAgent(a, events)
567 a.lifecycleMu.Lock()
568 a.updatedAt = summaryUpdatedAt
569 a.lastActivity = summaryLastActivity
570 a.lifecycleMu.Unlock()
571 }
572 a.mu.Lock()
573 a.transcriptLoaded = true
574 a.mu.Unlock()
575 return nil
576}
577
578// replayEventsIntoAgent re-derives an agent's transcript and last
579// known status hint from its event log. Mirrors applyEventToSink in
580// runner.go but writes directly to the Agent fields (no Sink because
581// the agent isn't being driven by a runner yet).
582//
583// Status precedence: explicit lifecycle events (agent_stopped) win
584// over inferred ones (assistant_message → idle). If the log contains
585// no terminator we keep status=StatusDetached so the user can resume.
586func resumePromptAcknowledged(events []Event, acceptedAt time.Time) bool {
587 if acceptedAt.IsZero() {
588 return false
589 }
590 for _, ev := range events {
591 if (ev.Type == EventTurnStarted || ev.Type == "turn_start") && !ev.Time.Before(acceptedAt) {
592 return true
593 }
594 }
595 return false
596}
597
598func replayEventsIntoAgent(a *Agent, evs []Event) {
599 terminal := false
600 for _, ev := range evs {
601 replayEventTranscript(a, ev)
602 switch ev.Type {
603 case EventTurnStarted, "turn_start":
604 turnID := ev.TurnID
605 if turnID == "" {
606 if step, ok := ev.Data["step"].(float64); ok {
607 turnID = fmt.Sprintf("turn-%d", int(step))
608 }
609 }
610 a.setTurnState(TurnRunning, turnID)
611 case EventTurnResult, "turn_result":
612 if result, err := decodeTurnResultEvent(ev, a.ID, a.maxOutputBytes, a.maxOutputLines); err == nil {
613 a.setResult(result)
614 switch result.Status {
615 case ResultFailed:
616 a.setTurnState(TurnFailed, result.TurnID)
617 case ResultCanceled:
618 a.setTurnState(TurnCanceled, result.TurnID)
619 default:
620 a.setTurnState(TurnSucceeded, result.TurnID)
621 }
622 }
623 case EventAgentIdle, "agent_idle":
624 a.setTurnState(TurnIdle, ev.TurnID)
625 case EventAgentExited, "agent_exited":
626 terminal = true
627 a.mu.Lock()
628 a.status = StatusDone
629 a.activity = "done (offline)"
630 a.mu.Unlock()
631 case "agent_stopped":
632 terminal = true
633 reason, _ := ev.Data["reason"].(string)
634 a.mu.Lock()
635 switch reason {
636 case "cancelled":
637 a.status = StatusKilled
638 a.activity = "cancelled (offline)"
639 case "shutdown":
640 a.status = StatusDone
641 a.activity = "shutdown (offline)"
642 case "exit":
643 if code, ok := ev.Data["code"].(float64); ok && code != 0 {
644 a.status = StatusFailed
645 a.activity = fmt.Sprintf("exit %d (offline)", int(code))
646 } else {
647 a.status = StatusDone
648 a.activity = "done (offline)"
649 }
650 default:
651 a.status = StatusDone
652 a.activity = "stopped (offline)"
653 }
654 a.mu.Unlock()
655 }
656 }
657 if !terminal {
658 // Non-terminal log means the previous parent died mid-run.
659 // Leave status=StatusDetached but record a hint so the
660 // dashboard shows something useful.
661 a.mu.Lock()
662 if a.activity == "detached" && len(a.transcript) > 0 {
663 a.activity = "detached (resume to continue)"
664 }
665 a.mu.Unlock()
666 }
667}
668
669// replayTranscriptIntoAgent rebuilds only the visible transcript. Newer
670// metadata already contains the authoritative dashboard lifecycle summary,
671// so opening history must not overwrite it with an earlier event attempt.
672func replayTranscriptIntoAgent(a *Agent, evs []Event) {
673 for _, ev := range evs {
674 replayEventTranscript(a, ev)
675 }
676}
677
678func replayEventTranscript(a *Agent, ev Event) {
679 if isAssistantStreamBoundary(ev.Type) {
680 a.resetStreamingAssistant()
681 }
682 switch ev.Type {
683 case "message.delta":
684 if delta, _ := ev.Data["delta"].(string); delta != "" {
685 a.appendAssistantDelta(delta)
686 }
687 case "assistant_message", "user_message":
688 var text []string
689 if c, ok := ev.Data["content"].([]any); ok {
690 for _, blk := range c {
691 m, _ := blk.(map[string]any)
692 if t, _ := m["type"].(string); t == "text" {
693 if txt, _ := m["text"].(string); txt != "" {
694 text = append(text, txt)
695 }
696 }
697 }
698 }
699 message := strings.Join(text, "\n")
700 if ev.Type == "assistant_message" {
701 a.appendAssistantMessage(message)
702 } else {
703 a.appendUserMessage(message)
704 }
705 case "stdout":
706 if txt, _ := ev.Data["text"].(string); txt != "" {
707 a.appendTranscript(txt)
708 }
709 case "stderr":
710 if txt, _ := ev.Data["text"].(string); txt != "" {
711 a.appendTranscript("stderr: " + txt)
712 }
713 case "error":
714 if msg, _ := ev.Data["message"].(string); msg != "" {
715 a.appendTranscript("error: " + msg)
716 }
717 }
718}
719
720// Resume re-attaches a Runner to a previously-spawned agent. Durable
721// session and event files are kept, while the transient inbox path is
722// recalculated for the current runtime environment. Use this to
723// continue a subagent session across zut restarts:
724//
725// subagentMgr.Reload()
726// a, err := subagentMgr.Resume(ctx, "alpha-12345")
727// subagentMgr.SendUserTurn(a.ID, "where were we?")
728//
729// The agent must be in a non-running state (Detached, Done, Failed,
730// Killed). Resuming a still-running agent returns an error so two
731// runners don't race for the same session.
732func (f *Supervisor) Resume(ctx context.Context, id string) (*Agent, error) {
733 return f.resume(ctx, id, true, "")
734}
735
736// ResumeSession continues the existing child session without replaying the
737// original task. It is the explicit canonical spelling for Resume.
738func (f *Supervisor) ResumeSession(ctx context.Context, id string) (*Agent, error) {
739 return f.resume(ctx, id, true, "")
740}
741
742// ResumeWithPrompt continues an agent with a manager follow-up while retaining
743// its existing session. An idle live worker receives the turn through its
744// inbox; a terminal worker is restarted with the follow-up as its initial
745// prompt, avoiding a race with inbox listener startup.
746func (f *Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (*Agent, error) {
747 if strings.TrimSpace(prompt) == "" {
748 return nil, fmt.Errorf("subagent: follow-up prompt is required")
749 }
750
751 f.operationMu.Lock()
752 existing := f.Get(id)
753 if existing == nil {
754 f.operationMu.Unlock()
755 return nil, fmt.Errorf("subagent: no such agent %q", id)
756 }
757 status := existing.Status()
758 if status == StatusRunning {
759 if existing.ProcessState() != ProcessAlive || existing.TurnState() != TurnIdle {
760 f.operationMu.Unlock()
761 return nil, fmt.Errorf("subagent: agent %s is still working; wait for it to become idle before sending a follow-up", existing.ID)
762 }
763 if existing.inbox == nil {
764 f.operationMu.Unlock()
765 return nil, fmt.Errorf("subagent: agent %s has no inbox", existing.ID)
766 }
767 turnID := existing.CurrentTurnID()
768 // Reserve the worker before writing to its inbox. A second manager call
769 // must not enqueue another turn in the small interval before the worker
770 // reports its turn.start event. Persist the prompt first so a host exit
771 // before that event can resume and redeliver it.
772 previousPrompt, previousPromptAt := existing.setResumePrompt(prompt, f.cfg.Now())
773 existing.setTurnState(TurnQueued, turnID)
774 inbox := existing.inbox
775 command := NewCommand(CommandTurnStart, existing.ID, turnID, TurnStartPayload{Prompt: prompt})
776 f.persistAgent(existing)
777 f.operationMu.Unlock()
778
779 if err := inbox.SendCommandContext(ctx, command); err != nil {
780 f.operationMu.Lock()
781 if existing.Status() == StatusRunning && existing.TurnState() == TurnQueued && existing.CurrentTurnID() == turnID {
782 existing.setTurnState(TurnIdle, turnID)
783 existing.setResumePrompt(previousPrompt, previousPromptAt)
784 f.persistAgent(existing)
785 }
786 f.operationMu.Unlock()
787 return nil, fmt.Errorf("subagent: send follow-up to %s: %w", existing.ID, err)
788 }
789 return existing, nil
790 }
791 f.operationMu.Unlock()
792
793 return f.resume(ctx, id, true, prompt)
794}
795
796// RestartTask intentionally runs the stored original task again in a fresh
797// worker attempt. Callers should use ResumeSession for normal recovery.
798func (f *Supervisor) RestartTask(ctx context.Context, id string) (*Agent, error) {
799 return f.resume(ctx, id, false, "")
800}
801
802func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resumePrompt string) (*Agent, error) {
803 f.operationMu.Lock()
804 defer f.operationMu.Unlock()
805 if ctx == nil {
806 ctx = context.Background()
807 }
808 existing := f.Get(id)
809 if existing == nil {
810 return nil, fmt.Errorf("subagent: no such agent %q", id)
811 }
812 existing.mu.Lock()
813 st := existing.status
814 existing.mu.Unlock()
815 if st == StatusRunning || st == StatusPending {
816 return nil, fmt.Errorf("subagent: agent %s is still %s; stop it first", existing.ID, st)
817 }
818 select {
819 case <-existing.done:
820 default:
821 return nil, fmt.Errorf("subagent: agent %s is still finalizing; wait before resuming", existing.ID)
822 }
823 // A reloaded worker may still be alive even though this supervisor has no
824 // in-process Runner (the old host can disappear after the child daemon
825 // starts). Fence resume before touching the shared session/worktree: a
826 // second worker must never own the same inbox and files concurrently.
827 if inboxLive(existing.InboxPath) {
828 return nil, fmt.Errorf("subagent: agent %s still has a live worker; send shutdown or stop it first", existing.ID)
829 }
830 // A fresh Reload does not hydrate its event log. Before replaying a pending
831 // follow-up, hydrate it to see whether the child already acknowledged the
832 // turn before the previous host exited.
833 if resuming && existing.resumePrompt() != "" {
834 if err := existing.loadTranscript(); err != nil {
835 return nil, fmt.Errorf("subagent: replay pending follow-up: %w", err)
836 }
837 }
838 stateDir := existing.stateDirectory(f.cfg.Root)
839 lease, err := acquireAgentLease(stateDir)
840 if err != nil {
841 return nil, fmt.Errorf("subagents resume lease: %w", err)
842 }
843 leaseOwned := true
844 defer func() {
845 if leaseOwned {
846 _ = lease.Close()
847 }
848 }()
849 existingSnapshot := existing.Snapshot()
850
851 // Rebuild from the meta record so we don't carry stale runner
852 // state from a previous incarnation. The inbox is transient and
853 // may have been persisted under an incompatible filesystem by an
854 // older zut version, so always select it again on resume.
855 inboxPath, err := inboxSocketPath(f.cfg.Root, existing.ID)
856 if err != nil {
857 return nil, fmt.Errorf("subagent inbox path: %w", err)
858 }
859 now := f.cfg.Now()
860 fastMode := existing.FastMode
861 pendingResumePrompt, pendingResumePromptAt := existing.ResumePromptInfo()
862 if resumePrompt != "" {
863 pendingResumePrompt = resumePrompt
864 pendingResumePromptAt = now
865 }
866 m := agentMeta{
867 ID: existing.ID, Task: existing.Task,
868 OriginalTask: existing.OriginalTask,
869 Dir: existing.Dir, RepositoryRoot: existing.RepositoryRoot, Started: existing.Started,
870 ParentID: existing.ParentID, BatchID: existing.BatchID, RootSessionID: existing.RootSessionID,
871 Model: existing.Model, Provider: existing.Provider,
872 BaseURL: existing.BaseURL, InsecureTLS: existing.InsecureTLS,
873 Reasoning: existing.Reasoning, FastMode: &fastMode,
874 Subagent: existing.Subagent,
875 WorkspaceMode: existing.WorkspaceMode, WorkspacePath: existing.WorkspacePath,
876 WorkspaceBase: existing.WorkspaceBase, WorkspaceCapture: existing.WorkspaceCapture,
877 MaxTurns: existing.MaxTurns, Timeout: existing.Timeout, Tools: append([]string(nil), existing.Tools...),
878 WebSearchPolicy: childWebSearchPolicy(existing.WebSearchPolicy, existing.Subagent, existing.Tools),
879 CurrentTurnID: existingSnapshot.CurrentTurnID, Attempt: existingSnapshot.Attempt,
880 SessionID: existing.SessionID,
881 InboxPath: inboxPath, EventLogPath: existing.EventLogPath,
882 SessionPath: existing.SessionPath,
883 ResumePrompt: pendingResumePrompt,
884 ResumePromptAt: pendingResumePromptAt,
885 }
886
887 if len(m.Tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
888 m.Tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
889 }
890 if err := f.validateSpawnOptions(SpawnRequest{MaxTurns: m.MaxTurns, Tools: m.Tools}); err != nil {
891 return nil, err
892 }
893 workspaceMode := m.WorkspaceMode
894 if workspaceMode == "" {
895 workspaceMode = WorkspaceShared
896 }
897 if workspaceMode != WorkspaceShared && workspaceMode != WorkspaceWorktree {
898 return nil, fmt.Errorf("subagents: unknown workspace mode %q", workspaceMode)
899 }
900 if m.WorkspaceCapture != "" && m.WorkspaceCapture != CapturePatch && m.WorkspaceCapture != CaptureDiff {
901 return nil, fmt.Errorf("subagents: unknown workspace capture mode %q", m.WorkspaceCapture)
902 }
903 f.mu.Lock()
904 allowedRoots := append([]string(nil), f.cfg.Policy.AllowedRoots...)
905 f.mu.Unlock()
906 repositoryRoot := firstNonEmpty(m.RepositoryRoot, m.Dir, f.cfg.RepoRoot)
907 if err := validateWorkspaceRoot(repositoryRoot, workspaceMode, allowedRoots); err != nil {
908 return nil, err
909 }
910 maxTurns := m.MaxTurns
911 if maxTurns <= 0 {
912 maxTurns = f.cfg.Policy.MaxTurns
913 }
914 existingWorkspacePath := ""
915 if workspaceMode == WorkspaceWorktree {
916 existingWorkspacePath = m.WorkspacePath
917 }
918 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
919 Mode: workspaceMode,
920 RepositoryRoot: repositoryRoot,
921 StateDir: stateDir,
922 AgentID: existing.ID,
923 Base: m.WorkspaceBase,
924 Capture: m.WorkspaceCapture,
925 AllowedRoots: allowedRoots,
926 ExistingPath: existingWorkspacePath,
927 })
928 if err != nil {
929 return nil, err
930 }
931 m.Dir = workspace.Dir()
932 m.RepositoryRoot = workspace.RepositoryRoot()
933 m.WorkspacePath = workspace.Dir()
934 m.WorkspaceMode = workspace.Mode()
935 timeout := m.Timeout
936 if timeout <= 0 {
937 timeout = f.cfg.Policy.DefaultTimeout
938 }
939 runCtx, cancel := f.workerContext(timeout)
940 a := &Agent{
941 ID: m.ID,
942 Task: m.Task,
943 OriginalTask: firstNonEmpty(m.OriginalTask, m.Task),
944 Dir: m.Dir,
945 RepositoryRoot: firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot),
946 Started: m.Started,
947 ParentID: m.ParentID,
948 BatchID: m.BatchID,
949 RootSessionID: m.RootSessionID,
950 Model: m.Model,
951 Provider: m.Provider,
952 BaseURL: m.BaseURL,
953 InsecureTLS: m.InsecureTLS,
954 Reasoning: m.Reasoning,
955 FastMode: fastMode,
956 Subagent: m.Subagent,
957 SessionID: m.SessionID,
958 WorkspaceMode: m.WorkspaceMode,
959 WorkspacePath: m.WorkspacePath,
960 WorkspaceBase: m.WorkspaceBase,
961 WorkspaceCapture: m.WorkspaceCapture,
962 MaxTurns: maxTurns,
963 Timeout: timeout,
964 Attempt: m.Attempt,
965 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
966 Tools: append([]string(nil), m.Tools...),
967 WebSearchPolicy: childWebSearchPolicy(m.WebSearchPolicy, m.Subagent, m.Tools),
968 InboxPath: m.InboxPath,
969 EventLogPath: m.EventLogPath,
970 SessionPath: m.SessionPath,
971 Resuming: resuming,
972 resumePromptText: pendingResumePrompt,
973 resumePromptAt: pendingResumePromptAt,
974 inbox: NewInbox(m.InboxPath),
975 status: StatusPending,
976 activity: "resuming",
977 transcriptLoaded: true,
978 processState: ProcessPending,
979 turnState: TurnQueued,
980 updatedAt: now,
981 lastActivity: now,
982 currentTurnID: m.CurrentTurnID,
983 stateDir: stateDir,
984 lease: lease,
985 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
986 maxOutputLines: f.cfg.Policy.MaxOutputLines,
987 done: make(chan struct{}),
988 turnResults: make(chan *TurnResult, 16),
989 }
990 // Carry the previous transcript forward so the dashboard doesn't
991 // flash empty between resume and the first new event.
992 prev := existing.Transcript()
993 if len(prev) > 0 {
994 a.appendTranscript(strings.Join(prev, "\n"))
995 }
996 a.ctx, a.cancel = runCtx, cancel
997 a.persistFn = f.persistAgent
998 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
999 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
1000 a.runner = f.cfg.NewRunner(a)
1001 if err := writeAgentMeta(a.stateDirectory(f.cfg.Root), a); err != nil {
1002 if a.cancel != nil {
1003 a.cancel()
1004 }
1005 _ = workspace.Cleanup(context.Background())
1006 return nil, fmt.Errorf("subagents resume metadata: %w", err)
1007 }
1008
1009 f.mu.Lock()
1010 f.agents[a.ID] = a
1011 f.queue = append(f.queue, a)
1012 // Keep the agent's slot in f.order; replacing in-place avoids
1013 // reshuffling the dashboard's row ordering on resume.
1014 found := false
1015 for _, k := range f.order {
1016 if k == a.ID {
1017 found = true
1018 break
1019 }
1020 }
1021 if !found {
1022 f.order = append(f.order, a.ID)
1023 }
1024 f.mu.Unlock()
1025 leaseOwned = false
1026
1027 // Refreshing metadata happened before queue admission, so a later
1028 // supervisor can reconstruct this resumed attempt even if the runner
1029 // has not reached its first event yet.
1030 f.armQueueTimeout(a)
1031 f.schedule()
1032 return a, nil
1033}