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