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 "github.com/google/uuid"
27)
28
29// agentMeta is the durable identity record for one agent. Only fields
30// the supervisor needs to rebuild an Agent after a restart live here.
31// Adding a field leaves it zero when absent; removing or renaming a field
32// changes the persisted contract.
33//
34// Historical fields like `branch` and `isolated` are silently dropped
35// by encoding/json's permissive decoder when an older meta.json is
36// loaded; we don't need to keep them in the struct.
37type requirementMeta = RequirementSnapshot
38
39type agentMeta struct {
40 ID string `json:"id"`
41 Task string `json:"task"`
42 OriginalTask string `json:"original_task,omitempty"`
43 Dir string `json:"dir"`
44 RepositoryRoot string `json:"repository_root,omitempty"`
45 Started time.Time `json:"started"`
46 ParentID string `json:"parent_id,omitempty"`
47 BatchID string `json:"batch_id,omitempty"`
48 RootSessionID string `json:"root_session_id,omitempty"`
49 Model string `json:"model,omitempty"`
50 Provider string `json:"provider,omitempty"`
51 BaseURL string `json:"base_url,omitempty"`
52 InsecureTLS bool `json:"insecure_tls,omitempty"`
53 Reasoning string `json:"reasoning,omitempty"`
54 FastMode *bool `json:"fast_mode,omitempty"`
55 Subagent string `json:"subagent,omitempty"`
56 WorkspaceMode WorkspaceMode `json:"workspace_mode,omitempty"`
57 WorkspacePath string `json:"workspace_path,omitempty"`
58 WorkspaceBase string `json:"workspace_base,omitempty"`
59 WorkspaceCapture CaptureMode `json:"workspace_capture,omitempty"`
60 MaxTurns int `json:"max_turns,omitempty"`
61 LifetimeTurns int `json:"lifetime_turns,omitempty"`
62 CurrentRunTurns int `json:"current_run_turns,omitempty"`
63 Timeout time.Duration `json:"timeout,omitempty"`
64 Tools []string `json:"tools,omitempty"`
65 WebSearchPolicy WebSearchPolicy `json:"web_search_policy,omitempty"`
66 Status Status `json:"status,omitempty"`
67 Activity string `json:"activity,omitempty"`
68 Finished time.Time `json:"finished,omitempty"`
69 Error string `json:"error,omitempty"`
70 ProcessState ProcessState `json:"process_state,omitempty"`
71 TurnState TurnState `json:"turn_state,omitempty"`
72 CurrentTurnID string `json:"current_turn_id,omitempty"`
73 Attempt int `json:"attempt,omitempty"`
74 ProcessPID int `json:"process_pid,omitempty"`
75 UpdatedAt time.Time `json:"updated_at,omitempty"`
76 LastActivity time.Time `json:"last_activity,omitempty"`
77 ResultRef string `json:"result_ref,omitempty"`
78 Requirement requirementMeta `json:"requirement,omitempty"`
79 PatchRef string `json:"patch_ref,omitempty"`
80 ChangedFiles []string `json:"changed_files,omitempty"`
81 InboxPath string `json:"inbox_path"`
82 EventLogPath string `json:"event_log_path"`
83 SessionPath string `json:"session_path"`
84 // ResumePrompt is a follow-up accepted by subagent_resume but not yet
85 // acknowledged by the child with a turn.started event. It is durable so a
86 // host exit while the worker is queued does not lose the manager signal.
87 ResumePrompt string `json:"resume_prompt,omitempty"`
88 ResumePromptAt time.Time `json:"resume_prompt_at,omitempty"`
89 ResumePromptID string `json:"resume_prompt_id,omitempty"`
90 ResumeQueue []queuedResumePrompt `json:"resume_queue,omitempty"`
91
92 // SessionID, when non-empty, scopes the agent to a particular
93 // host zut session: the dashboard only shows agents whose
94 // SessionID matches the active session. Older meta.json files
95 // (and agents spawned outside of any session, e.g. by tests or
96 // scripted callers that didn't call SetActiveSession) have an
97 // empty SessionID and are visible from every session as a
98 // unscoped-agent fallback.
99 SessionID string `json:"session_id,omitempty"`
100}
101
102func metaPath(stateDir string) string { return filepath.Join(stateDir, "meta.json") }
103
104// writeAgentMeta serialises a's identity into stateDir/meta.json. The
105// write is atomic (tmp + rename) so a crash mid-write can't leave a
106// half-parsable file that fails Reload.
107func writeAgentMeta(stateDir string, a *Agent) error {
108 if a == nil {
109 return errors.New("subagent meta: nil agent")
110 }
111 a.persistMu.Lock()
112 defer a.persistMu.Unlock()
113
114 fastMode := a.FastMode
115 webSearchPolicy := childWebSearchPolicy(a.WebSearchPolicy, a.Subagent, a.Tools)
116 a.mu.Lock()
117 status := a.status
118 activity := a.activity
119 finished := a.finished
120 resumePrompt := a.resumePromptText
121 resumePromptAt := a.resumePromptAt
122 resumePromptID := a.resumePromptID
123 resumeQueue := append([]queuedResumePrompt(nil), a.resumePromptQueue...)
124 errText := ""
125 if a.lastErr != nil {
126 errText = a.lastErr.Error()
127 }
128 a.mu.Unlock()
129 a.lifecycleMu.Lock()
130 processState := a.processState
131 turnState := a.turnState
132 currentTurnID := a.currentTurnID
133 lifetimeTurns := a.LifetimeTurns
134 currentRunTurns := a.CurrentRunTurns
135 attempt := a.Attempt
136 processPID := a.ProcessPID
137 updatedAt := a.updatedAt
138 lastActivity := a.lastActivity
139 resultRef := a.resultRef
140 requirement := a.requirement
141 patchRef := a.patchRef
142 changedFiles := append([]string(nil), a.changedFiles...)
143 a.lifecycleMu.Unlock()
144 m := agentMeta{
145 ID: a.ID,
146 Task: a.Task,
147 OriginalTask: a.OriginalTask,
148 Dir: a.Dir,
149 RepositoryRoot: a.RepositoryRoot,
150 Started: a.Started,
151 ParentID: a.ParentID,
152 BatchID: a.BatchID,
153 RootSessionID: a.RootSessionID,
154 Model: a.Model,
155 Provider: a.Provider,
156 BaseURL: a.BaseURL,
157 InsecureTLS: a.InsecureTLS,
158 Reasoning: a.Reasoning,
159 FastMode: &fastMode,
160 Subagent: a.Subagent,
161 WorkspaceMode: a.WorkspaceMode,
162 WorkspacePath: a.WorkspacePath,
163 WorkspaceBase: a.WorkspaceBase,
164 WorkspaceCapture: a.WorkspaceCapture,
165 MaxTurns: a.MaxTurns,
166 LifetimeTurns: lifetimeTurns,
167 CurrentRunTurns: currentRunTurns,
168 Timeout: a.Timeout,
169 Tools: append([]string(nil), a.Tools...),
170 WebSearchPolicy: webSearchPolicy,
171 Status: status,
172 Activity: activity,
173 Finished: finished,
174 Error: errText,
175 ProcessState: processState,
176 TurnState: turnState,
177 CurrentTurnID: currentTurnID,
178 Attempt: attempt,
179 ProcessPID: processPID,
180 UpdatedAt: updatedAt,
181 LastActivity: lastActivity,
182 ResultRef: resultRef,
183 Requirement: requirement,
184 PatchRef: patchRef,
185 ChangedFiles: changedFiles,
186 InboxPath: a.InboxPath,
187 EventLogPath: a.EventLogPath,
188 SessionPath: a.SessionPath,
189 ResumePrompt: resumePrompt,
190 ResumePromptAt: resumePromptAt,
191 ResumePromptID: resumePromptID,
192 ResumeQueue: resumeQueue,
193 SessionID: a.SessionID,
194 }
195 b, err := json.MarshalIndent(m, "", " ")
196 if err != nil {
197 return fmt.Errorf("subagent meta marshal: %w", err)
198 }
199 if err := os.MkdirAll(stateDir, 0o700); err != nil {
200 return fmt.Errorf("subagents meta dir: %w", err)
201 }
202 final := metaPath(stateDir)
203 // Use a unique temporary file while holding the per-agent persistence
204 // lock. A fixed meta.json.tmp is vulnerable to a concurrent writer (or a
205 // stale file after a crash) replacing another writer's snapshot.
206 tmpFile, err := os.CreateTemp(stateDir, ".meta.json.tmp-*")
207 if err != nil {
208 return fmt.Errorf("subagents meta temp: %w", err)
209 }
210 tmp := tmpFile.Name()
211 removeTemp := true
212 defer func() {
213 _ = tmpFile.Close()
214 if removeTemp {
215 _ = os.Remove(tmp)
216 }
217 }()
218 if err := tmpFile.Chmod(0o600); err != nil {
219 return fmt.Errorf("subagents meta permissions: %w", err)
220 }
221 if _, err := tmpFile.Write(append(b, '\n')); err != nil {
222 return fmt.Errorf("subagents meta write: %w", err)
223 }
224 if err := tmpFile.Sync(); err != nil {
225 return fmt.Errorf("subagents meta sync: %w", err)
226 }
227 if err := tmpFile.Close(); err != nil {
228 return fmt.Errorf("subagents meta close: %w", err)
229 }
230 if err := os.Rename(tmp, final); err != nil {
231 return fmt.Errorf("subagent meta rename: %w", err)
232 }
233 removeTemp = false
234 // Sync the containing directory where the platform supports it so the
235 // rename itself survives a host crash, not just the temporary file bytes.
236 if err := syncDirectory(stateDir); err != nil {
237 return fmt.Errorf("subagents meta directory sync: %w", err)
238 }
239 return nil
240}
241
242// readAgentMeta loads one meta.json. Returns os.ErrNotExist (wrapped)
243// when the file is missing so callers can distinguish "no such agent"
244// from "corrupt metadata".
245func readAgentMeta(stateDir string) (agentMeta, error) {
246 var m agentMeta
247 b, err := os.ReadFile(metaPath(stateDir))
248 if err != nil {
249 return m, err
250 }
251 if err := json.Unmarshal(b, &m); err != nil {
252 return m, fmt.Errorf("subagent meta parse %s: %w", stateDir, err)
253 }
254 if m.ID == "" {
255 return m, fmt.Errorf("subagent meta %s: missing id", stateDir)
256 }
257 return m, nil
258}
259
260func safeAgentID(id string) bool {
261 return id != "" && id != "." && id != ".." && filepath.Base(id) == id && !strings.ContainsAny(id, `/\\`)
262}
263
264func (f *Supervisor) sanitizeReloadMeta(stateDir, root string, m *agentMeta) error {
265 if m == nil {
266 return errors.New("subagents reload: nil metadata")
267 }
268 if m.EventLogPath == "" {
269 m.EventLogPath = filepath.Join(stateDir, "events.jsonl")
270 }
271 if m.SessionPath == "" {
272 m.SessionPath = filepath.Join(stateDir, "session.json")
273 }
274 if filepath.Base(m.EventLogPath) != "events.jsonl" || !pathWithin(m.EventLogPath, stateDir) {
275 return fmt.Errorf("subagents reload %s: event log path escapes agent state", stateDir)
276 }
277 if filepath.Base(m.SessionPath) != "session.json" || !pathWithin(m.SessionPath, stateDir) {
278 return fmt.Errorf("subagents reload %s: session path escapes agent state", stateDir)
279 }
280 if m.InboxPath == "" || !validRuntimeInboxPath(m.InboxPath, m.ID) {
281 // Old metadata may point at a runtime directory from a previous
282 // environment. Recompute that transient path rather than ever
283 // dialing an arbitrary socket supplied by metadata.
284 expectedInbox, err := inboxSocketPath(root, m.ID)
285 if err != nil {
286 return fmt.Errorf("subagents reload %s: inbox path: %w", stateDir, err)
287 }
288 m.InboxPath = expectedInbox
289 }
290 return nil
291}
292
293func validRuntimeInboxPath(path, id string) bool {
294 if !filepath.IsAbs(path) || (filepath.Base(path) != id+".sock" && filepath.Base(path) != shortHash(id)+".sock") {
295 return false
296 }
297 dir := filepath.Dir(path)
298 if !strings.HasPrefix(filepath.Base(dir), "zut-subagents-") {
299 return false
300 }
301 bases := []string{os.TempDir(), "/tmp"}
302 if runtimeDir := os.Getenv("XDG_RUNTIME_DIR"); filepath.IsAbs(runtimeDir) {
303 bases = append(bases, runtimeDir)
304 }
305 for _, base := range bases {
306 if pathWithin(dir, filepath.Clean(base)) {
307 return true
308 }
309 }
310 return false
311}
312
313func firstNonEmpty(values ...string) string {
314 for _, value := range values {
315 if strings.TrimSpace(value) != "" {
316 return value
317 }
318 }
319 return ""
320}
321
322// Reload scans <root>/agents/*/meta.json and re-registers every
323// previously-spawned agent as a StatusDetached entry. Agents already
324// present in memory are left alone (Reload is idempotent and safe to
325// call after Spawn, though in practice the cli invokes it exactly
326// once just after New()).
327//
328// The reloaded agents have no live Runner; the user can:
329// - view their transcript (the dashboard reads from EventLogPath),
330// - resume them via Supervisor.Resume (starts a fresh subprocess on the
331// same worktree / session / inbox path),
332// - remove them (worktree + meta + events log gone).
333//
334// Reload returns the number of agents loaded plus any per-directory
335// error encountered. Malformed entries are skipped rather than
336// failing the whole reload — one bad meta.json shouldn't hide the
337// rest of the subagents.
338func (f *Supervisor) Reload() (loaded int, errs []error) {
339 root := filepath.Clean(f.cfg.Root)
340 if root == "." {
341 return 0, nil
342 }
343 loaded, errs = f.reloadRoot(root)
344 errs = append(errs, f.reloadBatches(root)...)
345 return loaded, errs
346}
347
348func (f *Supervisor) reloadRoot(root string) (loaded int, errs []error) {
349 agentsDir := filepath.Join(root, "agents")
350 entries, err := os.ReadDir(agentsDir)
351 if err != nil {
352 if errors.Is(err, os.ErrNotExist) {
353 return 0, nil
354 }
355 return 0, []error{fmt.Errorf("subagents reload %s: %w", root, err)}
356 }
357
358 // Sort by directory name so the load order is stable across runs.
359 names := make([]string, 0, len(entries))
360 for _, e := range entries {
361 if e.IsDir() {
362 names = append(names, e.Name())
363 }
364 }
365 sort.Strings(names)
366
367 for _, name := range names {
368 stateDir := filepath.Join(agentsDir, name)
369 m, err := readAgentMeta(stateDir)
370 if err != nil {
371 if errors.Is(err, os.ErrNotExist) {
372 continue
373 }
374 errs = append(errs, err)
375 continue
376 }
377 if filepath.Base(stateDir) != m.ID || !safeAgentID(m.ID) {
378 errs = append(errs, fmt.Errorf("subagents reload %s: unsafe or mismatched agent id", stateDir))
379 continue
380 }
381 if err := f.sanitizeReloadMeta(stateDir, root, &m); err != nil {
382 errs = append(errs, err)
383 continue
384 }
385
386 f.mu.Lock()
387 if _, exists := f.agents[m.ID]; exists {
388 f.mu.Unlock()
389 continue
390 }
391 a, requirementReconciled := f.buildDetachedAgent(m)
392 f.agents[m.ID] = a
393 f.order = append(f.order, m.ID)
394 f.mu.Unlock()
395 if requirementReconciled {
396 if err := f.persistAgent(a); err != nil {
397 errs = append(errs, fmt.Errorf("persist reconciled requirement %s: %w", a.ID, err))
398 }
399 }
400 loaded++
401 }
402 return loaded, errs
403}
404
405// buildDetachedAgent constructs an Agent from compact metadata with no running
406// Runner. Its transcript stays unloaded until a caller requests it, while the
407// persisted lifecycle summary keeps the dashboard immediately useful.
408//
409// The returned Agent has a closed `done` channel because Wait should return
410// instantly: there is nothing to wait for.
411func (f *Supervisor) buildDetachedAgent(m agentMeta) (*Agent, bool) {
412 // Metadata without a repository identity uses the current RepoRoot;
413 // records with one retain their persisted repository so a restart from
414 // another checkout cannot silently redirect the child.
415 dir := m.Dir
416 if m.WorkspaceMode != WorkspaceWorktree && m.RepositoryRoot == "" && f.cfg.RepoRoot != "" {
417 // Preserve the best-effort migration only for records without an
418 // explicit repository identity.
419 dir = f.cfg.RepoRoot
420 }
421 if dir == "" {
422 dir = firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot)
423 }
424 fastMode := f.cfg.FastMode
425 if m.FastMode != nil {
426 fastMode = *m.FastMode
427 }
428 turnState := m.TurnState
429 if turnState == "" {
430 turnState = TurnIdle
431 }
432 lastActivity := m.LastActivity
433 if lastActivity.IsZero() {
434 lastActivity = m.Started
435 }
436 persistedStatus := m.Status
437 status := StatusDetached
438 activity := strings.TrimSpace(m.Activity)
439 processState := m.ProcessState
440 legacyEventState := persistedStatus == ""
441 if legacyEventState {
442 activity = "detached"
443 processState = ProcessDetached
444 }
445 if processState == "" {
446 processState = ProcessDetached
447 }
448 if activity == "" {
449 switch persistedStatus {
450 case StatusDone:
451 activity = "done"
452 case StatusFailed:
453 activity = "failed"
454 case StatusKilled:
455 activity = "cancelled"
456 default:
457 activity = "detached"
458 }
459 }
460 if persistedStatus == StatusRunning || persistedStatus == StatusPending {
461 activity = "detached (resume to continue)"
462 processState = ProcessDetached
463 }
464 if inboxLive(m.InboxPath) {
465 processState = ProcessAlive
466 activity = "live (detached)"
467 }
468 var lastErr error
469 if m.Error != "" {
470 lastErr = errors.New(m.Error)
471 }
472 agentStateDir := f.agentStateDir(m.ID)
473 if m.EventLogPath != "" {
474 // EventLogPath is <state-dir>/events.jsonl; keep the reloaded
475 // agent's state scoped to its own directory. Using the parent
476 // agents/ directory here would make Remove delete sibling agents
477 // and would make resumed result/patch writes collide.
478 agentStateDir = filepath.Dir(m.EventLogPath)
479 }
480 a := &Agent{
481 ID: m.ID,
482 Task: m.Task,
483 OriginalTask: firstNonEmpty(m.OriginalTask, m.Task),
484 Dir: dir,
485 RepositoryRoot: firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot),
486 Started: m.Started,
487 ParentID: m.ParentID,
488 BatchID: m.BatchID,
489 RootSessionID: m.RootSessionID,
490 Model: m.Model,
491 Provider: m.Provider,
492 BaseURL: m.BaseURL,
493 InsecureTLS: m.InsecureTLS,
494 Reasoning: m.Reasoning,
495 FastMode: fastMode,
496 Subagent: m.Subagent,
497 WorkspaceMode: m.WorkspaceMode,
498 WorkspacePath: m.WorkspacePath,
499 WorkspaceBase: m.WorkspaceBase,
500 WorkspaceCapture: m.WorkspaceCapture,
501 MaxTurns: m.MaxTurns,
502 LifetimeTurns: m.LifetimeTurns,
503 CurrentRunTurns: m.CurrentRunTurns,
504 Timeout: m.Timeout,
505 Tools: append([]string(nil), m.Tools...),
506 WebSearchPolicy: childWebSearchPolicy(m.WebSearchPolicy, m.Subagent, m.Tools),
507 Attempt: m.Attempt,
508 ProcessPID: m.ProcessPID,
509 InboxPath: m.InboxPath,
510 EventLogPath: m.EventLogPath,
511 SessionPath: m.SessionPath,
512 resumePromptText: m.ResumePrompt,
513 resumePromptAt: m.ResumePromptAt,
514 resumePromptID: m.ResumePromptID,
515 resumePromptQueue: append([]queuedResumePrompt(nil), m.ResumeQueue...),
516 SessionID: m.SessionID,
517 inbox: NewInbox(m.InboxPath),
518 status: status,
519 activity: activity,
520 finished: m.Finished,
521 lastErr: lastErr,
522 legacyEventState: legacyEventState,
523 processState: processState,
524 turnState: turnState,
525 currentTurnID: m.CurrentTurnID,
526 updatedAt: m.UpdatedAt,
527 lastActivity: lastActivity,
528 stateDir: agentStateDir,
529 resultRef: m.ResultRef,
530 requirement: m.Requirement,
531 patchRef: m.PatchRef,
532 changedFiles: append([]string(nil), m.ChangedFiles...),
533 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
534 maxOutputLines: f.cfg.Policy.MaxOutputLines,
535 done: make(chan struct{}),
536 turnResults: make(chan *TurnResult, 16),
537 persistFn: f.persistAgent,
538 }
539 if a.updatedAt.IsZero() {
540 a.updatedAt = lastActivity
541 }
542 a.closeDone()
543
544 // The append-only log may contain a terminal event from an earlier
545 // attempt. A live socket is stronger evidence for the current process;
546 // keep the reloaded entry detached/live so Stop and Remove cannot treat a
547 // resumed worker as already finished.
548 if inboxLive(a.InboxPath) {
549 a.mu.Lock()
550 a.status = StatusDetached
551 a.activity = "live (detached)"
552 a.mu.Unlock()
553 a.setProcessState(ProcessAlive)
554 }
555 resultDir := f.agentStateDir(a.ID)
556 if a.EventLogPath != "" {
557 resultDir = filepath.Dir(a.EventLogPath)
558 }
559 beforeRequirement := a.requirementSnapshot()
560 if result, err := readTurnResult(resultDir); err == nil && validateTurnResultAgent(result, a.ID) == nil {
561 bounded := result.Bounded(f.cfg.Policy.MaxOutputBytes, f.cfg.Policy.MaxOutputLines)
562 a.setResult(bounded)
563 if requirementAcceptsResult(beforeRequirement, bounded) {
564 a.resolveRequirement(0, bounded, "", true)
565 }
566 }
567 if a.requirementSnapshot().pending() {
568 a.resolveRequirement(0, &TurnResult{
569 Status: ResultFailed,
570 Error: &ResultError{
571 Code: "required_outcome_unobserved",
572 Message: "host restarted before the required worker outcome was durably observed",
573 },
574 }, "", true)
575 }
576 return a, beforeRequirement != a.requirementSnapshot()
577}
578
579// LoadTranscript hydrates one reloaded agent's durable event history. Reload
580// deliberately skips this work so startup stays bounded by metadata size;
581// callers request it only when opening the transcript or otherwise reading it.
582func (f *Supervisor) LoadTranscript(id string) error {
583 a := f.Get(id)
584 if a == nil {
585 return fmt.Errorf("subagent: no such agent %q", id)
586 }
587 return a.loadTranscript()
588}
589
590func (a *Agent) loadTranscript() error {
591 if a == nil {
592 return nil
593 }
594 a.transcriptMu.Lock()
595 defer a.transcriptMu.Unlock()
596
597 a.mu.Lock()
598 if a.transcriptLoaded {
599 a.mu.Unlock()
600 return nil
601 }
602 path := a.EventLogPath
603 legacyEventState := a.legacyEventState
604 a.mu.Unlock()
605 var summaryUpdatedAt, summaryLastActivity time.Time
606 if !legacyEventState {
607 a.lifecycleMu.Lock()
608 summaryUpdatedAt = a.updatedAt
609 summaryLastActivity = a.lastActivity
610 a.lifecycleMu.Unlock()
611 }
612 if path == "" {
613 a.mu.Lock()
614 a.transcriptLoaded = true
615 a.mu.Unlock()
616 return nil
617 }
618
619 events, err := ReadEventLog(path)
620 if err != nil {
621 return fmt.Errorf("read subagent transcript: %w", err)
622 }
623 if prompt, acceptedAt := a.ResumePromptInfo(); prompt != "" && resumePromptAcknowledged(events, prompt, acceptedAt) {
624 a.clearResumePrompt()
625 }
626 a.mu.Lock()
627 if a.transcriptLoaded {
628 a.mu.Unlock()
629 return nil
630 }
631 a.transcript = nil
632 a.lastAssistant = ""
633 a.outputBytes = 0
634 a.outputLines = 0
635 a.outputTruncated = false
636 a.mu.Unlock()
637
638 if legacyEventState {
639 // Metadata written before the dashboard summary existed has no
640 // terminal status to trust. Preserve its historical event-derived
641 // behavior once the user explicitly opens its transcript.
642 replayEventsIntoAgent(a, events)
643 } else {
644 replayTranscriptIntoAgent(a, events)
645 a.lifecycleMu.Lock()
646 a.updatedAt = summaryUpdatedAt
647 a.lastActivity = summaryLastActivity
648 a.lifecycleMu.Unlock()
649 }
650 a.mu.Lock()
651 a.transcriptLoaded = true
652 a.mu.Unlock()
653 return nil
654}
655
656// replayEventsIntoAgent re-derives an agent's transcript and last
657// known status hint from its event log. Mirrors applyEventToSink in
658// runner.go but writes directly to the Agent fields (no Sink because
659// the agent isn't being driven by a runner yet).
660//
661// Status precedence: explicit lifecycle events (agent_stopped) win
662// over inferred ones (assistant_message → idle). If the log contains
663// no terminator we keep status=StatusDetached so the user can resume.
664func resumePromptAcknowledged(events []Event, prompt string, acceptedAt time.Time) bool {
665 if prompt == "" || acceptedAt.IsZero() {
666 return false
667 }
668 userDurable := false
669 for _, ev := range events {
670 if ev.Time.Before(acceptedAt) {
671 continue
672 }
673 if ev.Type == "user_message" && userMessageText(ev) == prompt {
674 // The worker emits this event only after the matching session
675 // message has been appended and flushed. A crash here leaves the
676 // prompt pending, which is intentionally safe to replay.
677 userDurable = true
678 continue
679 }
680 if userDurable && isDelegatedTurnStart(ev) {
681 return true
682 }
683 }
684 return false
685}
686
687func replayEventsIntoAgent(a *Agent, evs []Event) {
688 terminal := false
689 for _, ev := range evs {
690 replayEventTranscript(a, ev)
691 switch ev.Type {
692 case EventTurnStarted:
693 if !isDelegatedTurnStart(ev) {
694 continue
695 }
696 a.setTurnState(TurnRunning, eventTurnID(ev))
697 case EventTurnResult, "turn_result":
698 if result, err := decodeTurnResultEvent(ev, a.ID, a.maxOutputBytes, a.maxOutputLines); err == nil {
699 a.setResult(result)
700 switch result.Status {
701 case ResultFailed:
702 a.setTurnState(TurnFailed, result.TurnID)
703 case ResultCanceled:
704 a.setTurnState(TurnCanceled, result.TurnID)
705 default:
706 a.setTurnState(TurnSucceeded, result.TurnID)
707 }
708 }
709 case EventAgentIdle, "agent_idle":
710 a.setTurnState(TurnIdle, ev.TurnID)
711 case EventAgentExited, "agent_exited":
712 terminal = true
713 a.mu.Lock()
714 a.status = StatusDone
715 a.activity = "done (offline)"
716 a.mu.Unlock()
717 case "agent_stopped":
718 terminal = true
719 reason, _ := ev.Data["reason"].(string)
720 a.mu.Lock()
721 switch reason {
722 case "cancelled":
723 a.status = StatusKilled
724 a.activity = "cancelled (offline)"
725 case "shutdown":
726 a.status = StatusDone
727 a.activity = "shutdown (offline)"
728 case "exit":
729 if code, ok := ev.Data["code"].(float64); ok && code != 0 {
730 a.status = StatusFailed
731 a.activity = fmt.Sprintf("exit %d (offline)", int(code))
732 } else {
733 a.status = StatusDone
734 a.activity = "done (offline)"
735 }
736 default:
737 a.status = StatusDone
738 a.activity = "stopped (offline)"
739 }
740 a.mu.Unlock()
741 }
742 }
743 if !terminal {
744 // Non-terminal log means the previous parent died mid-run.
745 // Leave status=StatusDetached but record a hint so the
746 // dashboard shows something useful.
747 a.mu.Lock()
748 if a.activity == "detached" && len(a.transcript) > 0 {
749 a.activity = "detached (resume to continue)"
750 }
751 a.mu.Unlock()
752 }
753}
754
755// replayTranscriptIntoAgent rebuilds only the visible transcript. Newer
756// metadata already contains the authoritative dashboard lifecycle summary,
757// so opening history must not overwrite it with an earlier event attempt.
758func replayTranscriptIntoAgent(a *Agent, evs []Event) {
759 for _, ev := range evs {
760 replayEventTranscript(a, ev)
761 }
762}
763
764func replayEventTranscript(a *Agent, ev Event) {
765 if isAssistantStreamBoundary(ev.Type) {
766 a.resetStreamingAssistant()
767 }
768 switch ev.Type {
769 case "message.delta":
770 if delta, _ := ev.Data["delta"].(string); delta != "" {
771 a.appendAssistantDelta(delta)
772 }
773 case "assistant_message", "user_message":
774 var text []string
775 if c, ok := ev.Data["content"].([]any); ok {
776 for _, blk := range c {
777 m, _ := blk.(map[string]any)
778 if t, _ := m["type"].(string); t == "text" {
779 if txt, _ := m["text"].(string); txt != "" {
780 text = append(text, txt)
781 }
782 }
783 }
784 }
785 message := strings.Join(text, "\n")
786 if ev.Type == "assistant_message" {
787 a.appendAssistantMessage(message)
788 } else {
789 a.appendUserMessage(message)
790 }
791 case "stdout":
792 if txt, _ := ev.Data["text"].(string); txt != "" {
793 a.appendTranscript(txt)
794 }
795 case "stderr":
796 if txt, _ := ev.Data["text"].(string); txt != "" {
797 a.appendTranscript("stderr: " + txt)
798 }
799 case "error":
800 if msg, _ := ev.Data["message"].(string); msg != "" {
801 a.appendTranscript("error: " + msg)
802 }
803 }
804}
805
806// Resume re-attaches a Runner to a previously-spawned agent. Durable
807// session and event files are kept, while the transient inbox path is
808// recalculated for the current runtime environment. Use this to
809// continue a subagent session across zut restarts:
810//
811// subagentMgr.Reload()
812// a, err := subagentMgr.Resume(ctx, "alpha-12345")
813// subagentMgr.SendUserTurn(a.ID, "where were we?")
814//
815// The agent must be in a non-running state (Detached, Done, Failed,
816// Killed). Resuming a still-running agent returns an error so two
817// runners don't race for the same session.
818func (f *Supervisor) Resume(ctx context.Context, id string) (*Agent, error) {
819 return f.resumeWithHook(ctx, id, true, "", f.requiredRetryHook(id))
820}
821
822// ResumeSession continues the existing child session without replaying the
823// original task. It is the explicit canonical spelling for Resume.
824func (f *Supervisor) ResumeSession(ctx context.Context, id string) (*Agent, error) {
825 return f.resumeWithHook(ctx, id, true, "", f.requiredRetryHook(id))
826}
827
828// ResumeWithPrompt continues an agent with a manager follow-up while retaining
829// its existing session. An idle live worker receives the turn through its
830// inbox; an active live worker queues the follow-up for the next idle turn; a
831// terminal worker is restarted with the follow-up as its initial prompt,
832// avoiding a race with inbox listener startup.
833func (f *Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (*Agent, error) {
834 if hook := f.requiredRetryHook(id); hook != nil {
835 return f.ResumeWithPromptBefore(ctx, id, prompt, hook)
836 }
837 return f.ResumeWithPromptBefore(ctx, id, prompt, nil)
838}
839
840// ResumeRequiredWithPrompt retries required work. The next delegated turn
841// becomes the new durable target before it can be queued or delivered, so a
842// fast worker cannot finish between the resume call and requirement tracking.
843func (f *Supervisor) ResumeRequiredWithPrompt(ctx context.Context, id, prompt string) (*Agent, error) {
844 return f.ResumeRequiredWithPromptBefore(ctx, id, prompt, nil)
845}
846
847// ResumeRequiredWithPromptBefore combines durable requirement registration
848// with the host's pre-delivery callback.
849func (f *Supervisor) ResumeRequiredWithPromptBefore(ctx context.Context, id, prompt string, before func(*Agent, string) func()) (*Agent, error) {
850 return f.ResumeWithPromptBefore(ctx, id, prompt, combineRequiredRetryHook(before))
851}
852
853func (f *Supervisor) requiredRetryHook(id string) func(*Agent, string) func() {
854 a := f.Get(strings.TrimSpace(id))
855 if a == nil || !a.requirementSnapshot().Unmet() {
856 return nil
857 }
858 return combineRequiredRetryHook(nil)
859}
860
861func combineRequiredRetryHook(before func(*Agent, string) func()) func(*Agent, string) func() {
862 return func(a *Agent, prompt string) func() {
863 previous := a.prepareRequired(0)
864 var cleanup func()
865 if before != nil {
866 cleanup = before(a, prompt)
867 }
868 return func() {
869 if cleanup != nil {
870 cleanup()
871 }
872 a.restoreRequirement(previous)
873 }
874 }
875}
876
877// ResumeWithPromptBefore is ResumeWithPrompt with a pre-delivery hook. The
878// hook runs after the target Agent is known but before a queued or direct
879// follow-up is made deliverable. For a restarted worker it runs after the new
880// Agent is built and before its runner is scheduled. For queued and direct
881// delivery, the hook runs while operationMu is held and must not block or call
882// back into Supervisor methods. A non-nil cleanup is called when the operation
883// is rejected before delivery.
884func (f *Supervisor) ResumeWithPromptBefore(ctx context.Context, id, prompt string, before func(*Agent, string) func()) (*Agent, error) {
885 if ctx == nil {
886 ctx = context.Background()
887 }
888 if err := ctx.Err(); err != nil {
889 return nil, err
890 }
891 if strings.TrimSpace(prompt) == "" {
892 return nil, fmt.Errorf("subagent: follow-up prompt is required")
893 }
894
895 f.operationMu.Lock()
896 existing := f.Get(id)
897 if existing == nil {
898 f.operationMu.Unlock()
899 return nil, fmt.Errorf("subagent: no such agent %q", id)
900 }
901 status := existing.Status()
902 if status == StatusRunning {
903 if existing.ProcessState() != ProcessAlive {
904 f.operationMu.Unlock()
905 return nil, fmt.Errorf("subagent: agent %s is still working; wait for its worker to become available before sending a follow-up", existing.ID)
906 }
907 if existing.inbox == nil {
908 f.operationMu.Unlock()
909 return nil, fmt.Errorf("subagent: agent %s has no inbox", existing.ID)
910 }
911 if err := ctx.Err(); err != nil {
912 f.operationMu.Unlock()
913 return nil, err
914 }
915 if existing.TurnState() != TurnRunning && existing.hasQueuedResumePrompt() {
916 f.operationMu.Unlock()
917 return nil, fmt.Errorf("subagent: agent %s already has a follow-up pending delivery", existing.ID)
918 }
919 if existing.TurnState() == TurnRunning {
920 // Active turns cannot accept a command through the worker inbox yet.
921 // Keep the prompt durable and let the idle transition deliver it in
922 // order after the current message turn finishes.
923 var undo func()
924 if before != nil {
925 undo = before(existing, prompt)
926 }
927 acceptedAt := f.cfg.Now()
928 commandID := existing.queueResumePrompt(prompt, acceptedAt)
929 if err := f.persistAgent(existing); err != nil {
930 existing.removeQueuedResumePrompt(commandID)
931 if undo != nil {
932 undo()
933 }
934 f.operationMu.Unlock()
935 return nil, fmt.Errorf("subagent: persist queued follow-up for %s: %w", existing.ID, err)
936 }
937 f.operationMu.Unlock()
938 return existing, nil
939 }
940 if existing.TurnState() != TurnIdle {
941 f.operationMu.Unlock()
942 return nil, fmt.Errorf("subagent: agent %s is still working; wait for it to become idle before sending a follow-up", existing.ID)
943 }
944 turnID := existing.CurrentTurnID()
945 previousRunTurns := existing.resetCurrentRunTurns()
946 var undo func()
947 if before != nil {
948 undo = before(existing, prompt)
949 }
950 // Reserve the worker before writing to its inbox. A second manager call
951 // must not enqueue another turn in the small interval before the worker
952 // reports its turn.start event. Persist the prompt first so a host exit
953 // before that event can resume and redeliver it.
954 previousPrompt := existing.setResumePrompt(prompt, f.cfg.Now())
955 existing.setTurnState(TurnQueued, turnID)
956 inbox := existing.inbox
957 command := NewCommand(CommandTurnStart, existing.ID, turnID, TurnStartPayload{Prompt: prompt, NewRun: true})
958 command.MessageID = existing.resumePromptCommandID()
959 if err := f.persistAgent(existing); err != nil {
960 existing.setTurnState(TurnIdle, turnID)
961 existing.setCurrentRunTurns(previousRunTurns)
962 existing.restoreResumePrompt(previousPrompt)
963 if undo != nil {
964 undo()
965 }
966 f.operationMu.Unlock()
967 return nil, fmt.Errorf("subagent: persist follow-up for %s: %w", existing.ID, err)
968 }
969 f.operationMu.Unlock()
970
971 if err := inbox.SendCommandContext(ctx, command); err != nil {
972 retryUnknown := false
973 f.operationMu.Lock()
974 if existing.Status() == StatusRunning && existing.TurnState() == TurnQueued && existing.CurrentTurnID() == turnID {
975 existing.setTurnState(TurnIdle, turnID)
976 existing.setCurrentRunTurns(previousRunTurns)
977 if errors.Is(err, ErrDeliveryUnknown) {
978 existing.rejectActiveResumePrompt()
979 retryUnknown = true
980 } else {
981 existing.restoreResumePrompt(previousPrompt)
982 if undo != nil {
983 undo()
984 }
985 }
986 if persistErr := f.persistAgent(existing); persistErr != nil {
987 err = fmt.Errorf("%w; persist recovery: %v", err, persistErr)
988 }
989 }
990 f.operationMu.Unlock()
991 if retryUnknown {
992 // The command keeps its durable identity. A retry is safe whether
993 // the worker parsed the first write or only received a prefix.
994 f.dispatchQueuedResume(ctx, existing)
995 }
996 return nil, fmt.Errorf("subagent: send follow-up to %s: %w", existing.ID, err)
997 }
998 return existing, nil
999 }
1000 f.operationMu.Unlock()
1001
1002 return f.resumeWithHook(ctx, id, true, prompt, before)
1003}
1004
1005// dispatchQueuedResumeWithTimeout keeps the idle callback bounded. Unlike a
1006// manager request, an idle transition has no caller context to propagate.
1007func (f *Supervisor) dispatchQueuedResumeWithTimeout(a *Agent) {
1008 timeout := f.cfg.Policy.ReconnectTimeout
1009 if timeout <= 0 {
1010 timeout = 5 * time.Second
1011 }
1012 ctx, cancel := context.WithTimeout(context.Background(), timeout)
1013 defer cancel()
1014 f.dispatchQueuedResume(ctx, a)
1015}
1016
1017// dispatchQueuedResume delivers the oldest follow-up accepted during an
1018// active turn after the worker reports that its inbox can accept a new turn.
1019// The operation lock serializes this transition with manager resume calls and
1020// the agent's durable prompt state prevents a host exit from losing it.
1021func (f *Supervisor) dispatchQueuedResume(ctx context.Context, a *Agent) {
1022 if ctx == nil {
1023 ctx = context.Background()
1024 }
1025 if a == nil {
1026 return
1027 }
1028 f.operationMu.Lock()
1029 if a.Status() != StatusRunning || a.ProcessState() != ProcessAlive || a.TurnState() != TurnIdle || a.inbox == nil {
1030 f.operationMu.Unlock()
1031 return
1032 }
1033 prompt, acceptedAt, commandID, ok := a.startQueuedResumePrompt()
1034 if !ok {
1035 f.operationMu.Unlock()
1036 return
1037 }
1038 turnID := a.CurrentTurnID()
1039 previousRunTurns := a.resetCurrentRunTurns()
1040 a.setTurnState(TurnQueued, turnID)
1041 inbox := a.inbox
1042 command := NewCommand(CommandTurnStart, a.ID, turnID, TurnStartPayload{Prompt: prompt, NewRun: true})
1043 command.MessageID = commandID
1044 if err := f.persistAgent(a); err != nil {
1045 a.restoreResumePromptToQueue(prompt, acceptedAt)
1046 a.setCurrentRunTurns(previousRunTurns)
1047 a.setTurnState(TurnIdle, turnID)
1048 f.operationMu.Unlock()
1049 return
1050 }
1051 f.operationMu.Unlock()
1052
1053 if err := inbox.SendCommandContext(ctx, command); err == nil {
1054 return
1055 }
1056
1057 // A failed delivery must put the prompt back at the head of the queue.
1058 // Keep the worker idle so a later retry, or an explicit resume, can send
1059 // it without skipping any subsequent follow-ups.
1060 f.operationMu.Lock()
1061 if a.Status() == StatusRunning && a.ProcessState() == ProcessAlive && a.TurnState() == TurnQueued && a.CurrentTurnID() == turnID {
1062 if a.restoreResumePromptToQueue(prompt, acceptedAt) {
1063 a.setCurrentRunTurns(previousRunTurns)
1064 a.setTurnState(TurnIdle, turnID)
1065 if persistErr := f.persistAgent(a); persistErr != nil {
1066 // persistAgent records the failure on the agent. Keep the explicit
1067 // check here because this recovery write is the only durable copy
1068 // of the restored queue head.
1069 a.recordPersistenceError(persistErr)
1070 }
1071 }
1072 }
1073 f.operationMu.Unlock()
1074}
1075
1076// RestartTask intentionally runs the stored original task again in a fresh
1077// worker attempt. Callers should use ResumeSession for normal recovery.
1078func (f *Supervisor) RestartTask(ctx context.Context, id string) (*Agent, error) {
1079 return f.resumeWithHook(ctx, id, false, "", f.requiredRetryHook(id))
1080}
1081
1082func (f *Supervisor) resumeWithHook(ctx context.Context, id string, resuming bool, resumePrompt string, before func(*Agent, string) func()) (*Agent, error) {
1083 f.operationMu.Lock()
1084 defer f.operationMu.Unlock()
1085 if ctx == nil {
1086 ctx = context.Background()
1087 }
1088 existing := f.Get(id)
1089 if existing == nil {
1090 return nil, fmt.Errorf("subagent: no such agent %q", id)
1091 }
1092 existing.mu.Lock()
1093 st := existing.status
1094 existing.mu.Unlock()
1095 if st == StatusRunning || st == StatusPending {
1096 return nil, fmt.Errorf("subagent: agent %s is still %s; stop it first", existing.ID, st)
1097 }
1098 select {
1099 case <-existing.done:
1100 default:
1101 return nil, fmt.Errorf("subagent: agent %s is still finalizing; wait before resuming", existing.ID)
1102 }
1103 // A reloaded worker may still be alive even though this supervisor has no
1104 // in-process Runner (the old host can disappear after the child daemon
1105 // starts). Fence resume before touching the shared session/worktree: a
1106 // second worker must never own the same inbox and files concurrently.
1107 if inboxLive(existing.InboxPath) {
1108 return nil, fmt.Errorf("subagent: agent %s still has a live worker; send shutdown or stop it first", existing.ID)
1109 }
1110 // A fresh Reload does not hydrate its event log. Before replaying a pending
1111 // follow-up, hydrate it to see whether the child already acknowledged the
1112 // turn before the previous host exited.
1113 if resuming && existing.resumePrompt() != "" {
1114 if err := existing.loadTranscript(); err != nil {
1115 return nil, fmt.Errorf("subagent: replay pending follow-up: %w", err)
1116 }
1117 }
1118 stateDir := existing.stateDirectory(f.cfg.Root)
1119 lease, err := acquireAgentLease(stateDir)
1120 if err != nil {
1121 return nil, fmt.Errorf("subagents resume lease: %w", err)
1122 }
1123 leaseOwned := true
1124 defer func() {
1125 if leaseOwned {
1126 _ = lease.Close()
1127 }
1128 }()
1129 existingSnapshot := existing.Snapshot()
1130
1131 // Rebuild from the meta record so we don't carry stale runner
1132 // state from a previous incarnation. The inbox is transient and
1133 // may have been persisted under an incompatible filesystem by an
1134 // older zut version, so always select it again on resume.
1135 inboxPath, err := inboxSocketPath(f.cfg.Root, existing.ID)
1136 if err != nil {
1137 return nil, fmt.Errorf("subagent inbox path: %w", err)
1138 }
1139 now := f.cfg.Now()
1140 fastMode := existing.FastMode
1141 lifetimeTurns := existing.LifetimeTurnsValue()
1142 currentRunTurns := existing.CurrentRunTurnsValue()
1143 pendingResume := existing.resumePromptState()
1144 pendingResumePrompt, pendingResumePromptAt, pendingResumePromptID := pendingResume.Prompt, pendingResume.AcceptedAt, pendingResume.CommandID
1145 pendingResumeQueue := existing.resumePromptQueueSnapshot()
1146 if pendingResumePrompt != "" && pendingResumePromptID == "" {
1147 pendingResumePromptID = uuid.NewString()
1148 }
1149 if resuming && pendingResumePrompt == "" && len(pendingResumeQueue) > 0 {
1150 // A host can exit after accepting an active-worker follow-up but before
1151 // dispatching it. Promote the oldest durable prompt before accepting a
1152 // new one so scheduler and completion-tracker order stay aligned.
1153 queued := pendingResumeQueue[0]
1154 pendingResumePrompt = queued.Prompt
1155 pendingResumePromptAt = queued.AcceptedAt
1156 pendingResumePromptID = queued.CommandID
1157 if pendingResumePromptID == "" {
1158 pendingResumePromptID = uuid.NewString()
1159 }
1160 pendingResumeQueue = pendingResumeQueue[1:]
1161 }
1162 hadPendingResumePrompt := pendingResumePrompt != ""
1163 if resumePrompt != "" {
1164 currentRunTurns = 0
1165 if pendingResumePrompt == "" {
1166 pendingResumePrompt = resumePrompt
1167 pendingResumePromptAt = now
1168 pendingResumePromptID = uuid.NewString()
1169 } else {
1170 pendingResumeQueue = append(pendingResumeQueue, queuedResumePrompt{Prompt: resumePrompt, AcceptedAt: now, CommandID: uuid.NewString()})
1171 }
1172 }
1173 if resuming && pendingResumePrompt != "" {
1174 // A retained follow-up is an explicit new run, including a queued
1175 // follow-up recovered after a host restart. Preserve lifetime turns but
1176 // never carry the previous run's allowance into this one.
1177 currentRunTurns = 0
1178 }
1179 if !resuming {
1180 currentRunTurns = 0
1181 }
1182 m := agentMeta{
1183 ID: existing.ID, Task: existing.Task,
1184 OriginalTask: existing.OriginalTask,
1185 Dir: existing.Dir, RepositoryRoot: existing.RepositoryRoot, Started: existing.Started,
1186 ParentID: existing.ParentID, BatchID: existing.BatchID, RootSessionID: existing.RootSessionID,
1187 Model: existing.Model, Provider: existing.Provider,
1188 BaseURL: existing.BaseURL, InsecureTLS: existing.InsecureTLS,
1189 Reasoning: existing.Reasoning, FastMode: &fastMode,
1190 Subagent: existing.Subagent,
1191 WorkspaceMode: existing.WorkspaceMode, WorkspacePath: existing.WorkspacePath,
1192 WorkspaceBase: existing.WorkspaceBase, WorkspaceCapture: existing.WorkspaceCapture,
1193 MaxTurns: existing.MaxTurns, LifetimeTurns: lifetimeTurns, CurrentRunTurns: currentRunTurns, Timeout: existing.Timeout, Tools: append([]string(nil), existing.Tools...),
1194 WebSearchPolicy: childWebSearchPolicy(existing.WebSearchPolicy, existing.Subagent, existing.Tools),
1195 CurrentTurnID: existingSnapshot.CurrentTurnID, Attempt: existingSnapshot.Attempt,
1196 Requirement: existingSnapshot.Requirement,
1197 SessionID: existing.SessionID,
1198 InboxPath: inboxPath, EventLogPath: existing.EventLogPath,
1199 SessionPath: existing.SessionPath,
1200 ResumePrompt: pendingResumePrompt,
1201 ResumePromptAt: pendingResumePromptAt,
1202 ResumePromptID: pendingResumePromptID,
1203 ResumeQueue: pendingResumeQueue,
1204 }
1205
1206 if len(m.Tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
1207 m.Tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
1208 }
1209 if err := f.validateSpawnOptions(SpawnRequest{MaxTurns: m.MaxTurns, Tools: m.Tools}); err != nil {
1210 return nil, err
1211 }
1212 workspaceMode := m.WorkspaceMode
1213 if workspaceMode == "" {
1214 workspaceMode = WorkspaceShared
1215 }
1216 if workspaceMode != WorkspaceShared && workspaceMode != WorkspaceWorktree {
1217 return nil, fmt.Errorf("subagents: unknown workspace mode %q", workspaceMode)
1218 }
1219 if m.WorkspaceCapture != "" && m.WorkspaceCapture != CapturePatch && m.WorkspaceCapture != CaptureDiff {
1220 return nil, fmt.Errorf("subagents: unknown workspace capture mode %q", m.WorkspaceCapture)
1221 }
1222 f.mu.Lock()
1223 allowedRoots := append([]string(nil), f.cfg.Policy.AllowedRoots...)
1224 f.mu.Unlock()
1225 repositoryRoot := firstNonEmpty(m.RepositoryRoot, m.Dir, f.cfg.RepoRoot)
1226 if err := validateWorkspaceRoot(repositoryRoot, workspaceMode, allowedRoots); err != nil {
1227 return nil, err
1228 }
1229 maxTurns := m.MaxTurns
1230 if maxTurns <= 0 {
1231 maxTurns = f.cfg.Policy.MaxTurns
1232 }
1233 existingWorkspacePath := ""
1234 if workspaceMode == WorkspaceWorktree {
1235 existingWorkspacePath = m.WorkspacePath
1236 }
1237 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
1238 Mode: workspaceMode,
1239 RepositoryRoot: repositoryRoot,
1240 StateDir: stateDir,
1241 AgentID: existing.ID,
1242 Base: m.WorkspaceBase,
1243 Capture: m.WorkspaceCapture,
1244 AllowedRoots: allowedRoots,
1245 ExistingPath: existingWorkspacePath,
1246 })
1247 if err != nil {
1248 return nil, err
1249 }
1250 m.Dir = workspace.Dir()
1251 m.RepositoryRoot = workspace.RepositoryRoot()
1252 m.WorkspacePath = workspace.Dir()
1253 m.WorkspaceMode = workspace.Mode()
1254 timeout := m.Timeout
1255 if timeout <= 0 {
1256 timeout = f.cfg.Policy.DefaultTimeout
1257 }
1258 runCtx, cancel := f.workerContext()
1259 a := &Agent{
1260 ID: m.ID,
1261 Task: m.Task,
1262 OriginalTask: firstNonEmpty(m.OriginalTask, m.Task),
1263 Dir: m.Dir,
1264 RepositoryRoot: firstNonEmpty(m.RepositoryRoot, f.cfg.RepoRoot),
1265 Started: m.Started,
1266 ParentID: m.ParentID,
1267 BatchID: m.BatchID,
1268 RootSessionID: m.RootSessionID,
1269 Model: m.Model,
1270 Provider: m.Provider,
1271 BaseURL: m.BaseURL,
1272 InsecureTLS: m.InsecureTLS,
1273 Reasoning: m.Reasoning,
1274 FastMode: fastMode,
1275 Subagent: m.Subagent,
1276 SessionID: m.SessionID,
1277 WorkspaceMode: m.WorkspaceMode,
1278 WorkspacePath: m.WorkspacePath,
1279 WorkspaceBase: m.WorkspaceBase,
1280 WorkspaceCapture: m.WorkspaceCapture,
1281 MaxTurns: maxTurns,
1282 LifetimeTurns: m.LifetimeTurns,
1283 CurrentRunTurns: m.CurrentRunTurns,
1284 Timeout: timeout,
1285 Attempt: m.Attempt,
1286 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
1287 Tools: append([]string(nil), m.Tools...),
1288 WebSearchPolicy: childWebSearchPolicy(m.WebSearchPolicy, m.Subagent, m.Tools),
1289 InboxPath: m.InboxPath,
1290 EventLogPath: m.EventLogPath,
1291 SessionPath: m.SessionPath,
1292 Resuming: resuming,
1293 resumePromptText: pendingResumePrompt,
1294 resumePromptAt: pendingResumePromptAt,
1295 resumePromptID: m.ResumePromptID,
1296 resumePromptQueue: append([]queuedResumePrompt(nil), pendingResumeQueue...),
1297 inbox: NewInbox(m.InboxPath),
1298 status: StatusPending,
1299 activity: "resuming",
1300 transcriptLoaded: true,
1301 processState: ProcessPending,
1302 turnState: TurnQueued,
1303 updatedAt: now,
1304 lastActivity: now,
1305 currentTurnID: m.CurrentTurnID,
1306 requirement: m.Requirement,
1307 stateDir: stateDir,
1308 lease: lease,
1309 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
1310 maxOutputLines: f.cfg.Policy.MaxOutputLines,
1311 done: make(chan struct{}),
1312 turnResults: make(chan *TurnResult, 16),
1313 }
1314 // Carry the previous transcript forward so the dashboard doesn't
1315 // flash empty between resume and the first new event.
1316 prev := existing.Transcript()
1317 if len(prev) > 0 {
1318 a.appendTranscript(strings.Join(prev, "\n"))
1319 }
1320 a.ctx, a.cancel = runCtx, cancel
1321 a.persistFn = f.persistAgent
1322 a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
1323 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
1324 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
1325 a.runner = f.cfg.NewRunner(a)
1326 if err := writeAgentMeta(a.stateDirectory(f.cfg.Root), a); err != nil {
1327 if a.cancel != nil {
1328 a.cancel()
1329 }
1330 _ = workspace.Cleanup(context.Background())
1331 return nil, fmt.Errorf("subagents resume metadata: %w", err)
1332 }
1333
1334 f.mu.Lock()
1335 f.agents[a.ID] = a
1336 f.queue = append(f.queue, a)
1337 // Keep the agent's slot in f.order; replacing in-place avoids
1338 // reshuffling the dashboard's row ordering on resume.
1339 found := false
1340 for _, k := range f.order {
1341 if k == a.ID {
1342 found = true
1343 break
1344 }
1345 }
1346 if !found {
1347 f.order = append(f.order, a.ID)
1348 }
1349 f.mu.Unlock()
1350 leaseOwned = false
1351
1352 // Refreshing metadata happened before queue admission, so a later
1353 // supervisor can reconstruct this resumed attempt even if the runner
1354 // has not reached its first event yet. Register the turn before the
1355 // scheduler can start the worker and emit a fast turn_end. A durable pending
1356 // prompt is consumed before the newly requested prompt, so tracker
1357 // registrations must preserve that scheduler order.
1358 var beforeCleanups []func()
1359 if before != nil {
1360 register := func(prompt string) {
1361 if cleanup := before(a, prompt); cleanup != nil {
1362 beforeCleanups = append(beforeCleanups, cleanup)
1363 }
1364 }
1365 if hadPendingResumePrompt {
1366 register(pendingResumePrompt)
1367 }
1368 if resumePrompt != "" {
1369 register(resumePrompt)
1370 }
1371 if !hadPendingResumePrompt && resumePrompt == "" {
1372 register("")
1373 }
1374 if err := f.persistAgent(a); err != nil {
1375 a.recordPersistenceError(err)
1376 for idx := len(beforeCleanups) - 1; idx >= 0; idx-- {
1377 beforeCleanups[idx]()
1378 }
1379 f.mu.Lock()
1380 f.agents[a.ID] = existing
1381 for idx, queued := range f.queue {
1382 if queued == a {
1383 f.queue = append(f.queue[:idx], f.queue[idx+1:]...)
1384 break
1385 }
1386 }
1387 f.mu.Unlock()
1388 if restoreErr := f.persistAgent(existing); restoreErr != nil {
1389 existing.recordPersistenceError(restoreErr)
1390 }
1391 if a.cancel != nil {
1392 a.cancel()
1393 }
1394 _ = lease.Close()
1395 return nil, err
1396 }
1397 }
1398 f.armQueueTimeout(a)
1399 f.schedule()
1400 return a, nil
1401}