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