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