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