1// Package subagents implements zut's subagent supervisor.
2//
3// A Supervisor manages headless, long-lived zut subprocesses and keeps their
4// process/turn state, event logs, sessions, results, and optional worktree
5// artifacts durable across supervisor restarts.
6//
7// Shared workspaces preserve the historical behavior, while an explicit
8// worktree request gives a child an isolated checkout whose patch is captured
9// before cleanup. The Runner abstraction lets tests inject a fake process;
10// production uses `zut --subagent-worker ...`.
11package subagents
12
13import (
14 "context"
15 "errors"
16 "fmt"
17 "os"
18 "path/filepath"
19 "sort"
20 "strings"
21 "sync"
22 "sync/atomic"
23 "time"
24)
25
26// Status is the high-level lifecycle state of an Agent.
27type Status string
28
29const (
30 StatusPending Status = "pending" // created, not started yet
31 StatusRunning Status = "running" // Runner.Run is in flight
32 StatusDone Status = "done" // Runner.Run returned nil
33 StatusFailed Status = "failed" // Runner.Run returned an error
34 StatusKilled Status = "killed" // Stop() called before completion
35 StatusDetached Status = "detached" // reloaded from disk; no live runner
36)
37
38// Config configures a Supervisor.
39type Config struct {
40 // Context is the supervisor lifetime. Child workers use it as their
41 // parent context rather than inheriting an individual tool/model-turn
42 // context. A nil value means context.Background().
43 Context context.Context
44
45 // Root is the directory under which per-agent state files live.
46 // Typically <ZutHome>/subagents, but tests pass a tempdir.
47 Root string
48
49 // Policy owns concurrency, timeout, output, and path safety limits.
50 Policy SubagentPolicy
51
52 // RepoRoot is the repository used by spawned agents. Shared-mode children
53 // run here; worktree-mode children use an isolated checkout created from
54 // this root.
55 RepoRoot string
56
57 // Provider is the active host provider. It is used only to decide whether
58 // an explicit child provider may inherit the host endpoint settings.
59 Provider string
60
61 // FastMode is copied to every newly spawned child and persisted so
62 // child subprocesses use the same OpenAI service-tier setting.
63 FastMode bool
64
65 // WebSearchPolicy is the already-resolved capability of the parent
66 // session. Generic children may inherit it only when the subagent policy
67 // also permits web_search; named profiles resolve their own explicit
68 // tools opt-in at SpawnReq.
69 WebSearchPolicy WebSearchPolicy
70
71 // BaseURL and InsecureTLS are the effective provider connection
72 // settings inherited by newly spawned children. They are copied to
73 // each Agent and persisted so a reload/resume does not silently fall
74 // back to the child's independent provider configuration.
75 BaseURL string
76 InsecureTLS bool
77
78 // NewRunner produces the Runner for an Agent. If nil, the default
79 // `zut --subagent-worker ...` exec runner is used. Tests inject a fake
80 // here.
81 NewRunner func(a *Agent) Runner
82
83 // ResolveCredential resolves the credential inherited by a child
84 // process. The default runner transfers it over the child's stdin,
85 // keeping command-backed credentials out of argv and the environment.
86 ResolveCredential func(ctx context.Context, provider string) (Credential, error)
87
88 // Now is a clock seam for tests; defaults to time.Now.
89 Now func() time.Time
90}
91
92// Credential is the resolved authentication state inherited by a subagent child.
93type Credential struct {
94 Value string `json:"value"`
95 Method string `json:"method"`
96 AccountID string `json:"account_id,omitempty"`
97}
98
99// Runner executes one agent task. Run blocks until the task finishes,
100// is cancelled via ctx, or hits an unrecoverable error.
101//
102// Run should report progress by writing short human-readable strings
103// to the activity channel and final transcript text to transcript.
104// Both channels are non-blocking sinks owned by the Supervisor; if the
105// dashboard isn't reading, sends are dropped.
106type Runner interface {
107 Run(ctx context.Context, sink Sink) error
108}
109
110// Sink is how a Runner reports activity and transcript back to the
111// supervisor. All methods are safe to call from any goroutine and
112// never block.
113type Sink interface {
114 // Activity sets the one-line "what is this agent doing right now"
115 // string shown in the dashboard.
116 Activity(msg string)
117 // Transcript appends a chunk of agent output (typically a final
118 // assistant message) to the agent's running transcript.
119 Transcript(chunk string)
120}
121
122// Supervisor supervises a set of Agents.
123type Supervisor struct {
124 cfg Config
125
126 mu sync.Mutex
127 operationMu sync.Mutex
128 lifetimeCtx context.Context
129 lifetimeCancel context.CancelFunc
130 agents map[string]*Agent
131 order []string // creation order for stable listing
132 queue []*Agent
133 active int
134 activeByParent map[string]int
135 activeByBatch map[string]int
136 batches map[string]*Batch
137
138 // activeSession is the host session id the dashboard is
139 // currently scoped to. When non-empty, SnapshotAll filters out
140 // agents whose SessionID doesn't match (unscoped
141 // agents with SessionID == "" are always shown). Spawn stamps
142 // new agents with this value so they appear only in the session
143 // that created them. When empty (the default), the historical
144 // "show everything" behaviour is preserved — important for
145 // tests and any scripted use of the Supervisor that doesn't bother
146 // with sessions.
147 activeSession string
148}
149
150// New constructs a Supervisor from cfg. Missing config fields are filled
151// with defaults.
152func New(cfg Config) *Supervisor {
153 if cfg.Now == nil {
154 cfg.Now = time.Now
155 }
156 if cfg.NewRunner == nil {
157 cfg.NewRunner = func(a *Agent) Runner {
158 return &execRunner{agent: a, resolveCredential: cfg.ResolveCredential}
159 }
160 }
161 cfg.Policy.normalize()
162 lifetimeParent := cfg.Context
163 if lifetimeParent == nil {
164 lifetimeParent = context.Background()
165 }
166 lifetimeCtx, lifetimeCancel := context.WithCancel(lifetimeParent)
167 return &Supervisor{
168 cfg: cfg,
169 lifetimeCtx: lifetimeCtx,
170 lifetimeCancel: lifetimeCancel,
171 agents: map[string]*Agent{},
172 activeByParent: map[string]int{},
173 activeByBatch: map[string]int{},
174 batches: map[string]*Batch{},
175 }
176}
177
178// workerContext derives a child worker context from the supervisor lifetime.
179// A positive timeout limits the worker lifetime; otherwise it is cancellable
180// only when the supervisor shuts down or the agent is stopped.
181func (f *Supervisor) workerContext(timeout time.Duration) (context.Context, context.CancelFunc) {
182 base := f.lifetimeCtx
183 if base == nil {
184 base = context.Background()
185 }
186 if timeout > 0 {
187 return context.WithTimeout(base, timeout)
188 }
189 return context.WithCancel(base)
190}
191
192// SetRepoRoot updates the shared working directory for subsequently
193// spawned agents. Existing agents keep the directory they were started
194// with; callers use this when the host session changes cwd.
195func (f *Supervisor) SetRepoRoot(root string) {
196 f.mu.Lock()
197 defer f.mu.Unlock()
198 f.cfg.RepoRoot = root
199}
200
201// RepoRoot returns the working directory used by subsequently spawned
202// agents.
203func (f *Supervisor) RepoRoot() string {
204 f.mu.Lock()
205 defer f.mu.Unlock()
206 return f.cfg.RepoRoot
207}
208
209// MaxTurns returns the maximum prompt-level turns allowed for subsequently
210// spawned agents.
211func (f *Supervisor) MaxTurns() int {
212 f.mu.Lock()
213 defer f.mu.Unlock()
214 return f.cfg.Policy.MaxTurns
215}
216
217// SetFastMode updates the fast-mode setting for subsequently spawned
218// children. Existing agents keep the setting captured at their spawn
219// boundary so resumes remain deterministic.
220func (f *Supervisor) SetFastMode(enabled bool) {
221 f.mu.Lock()
222 f.cfg.FastMode = enabled
223 f.mu.Unlock()
224}
225
226// FastMode reports the setting used for subsequently spawned children.
227func (f *Supervisor) FastMode() bool {
228 f.mu.Lock()
229 defer f.mu.Unlock()
230 return f.cfg.FastMode
231}
232
233// SetWebSearchPolicy updates the parent capability used for subsequently
234// spawned generic children. Existing agents retain their persisted decision.
235func (f *Supervisor) SetWebSearchPolicy(policy WebSearchPolicy) {
236 f.mu.Lock()
237 f.cfg.WebSearchPolicy = policy
238 f.mu.Unlock()
239}
240
241// WebSearchPolicy reports the parent capability used for new generic
242// children.
243func (f *Supervisor) WebSearchPolicy() WebSearchPolicy {
244 f.mu.Lock()
245 defer f.mu.Unlock()
246 return f.cfg.WebSearchPolicy
247}
248
249// SetProvider updates the provider used by subsequently spawned children.
250// Existing agents keep their persisted provider selection.
251func (f *Supervisor) SetProvider(provider string) {
252 f.mu.Lock()
253 f.cfg.Provider = strings.TrimSpace(provider)
254 f.mu.Unlock()
255}
256
257// SetProviderSettings updates the effective connection settings used by
258// subsequently spawned children. Existing agents keep their persisted values
259// so a model/provider switch cannot silently change a running or resumed
260// worker's endpoint.
261func (f *Supervisor) SetProviderSettings(baseURL string, insecureTLS bool) {
262 f.mu.Lock()
263 f.cfg.BaseURL = strings.TrimSpace(baseURL)
264 f.cfg.InsecureTLS = insecureTLS
265 f.mu.Unlock()
266}
267
268// ProviderSettings returns the settings used by subsequently spawned
269// children. It is primarily useful to host integrations that need to verify
270// the manager follows the active provider selection.
271func (f *Supervisor) ProviderSettings() (string, bool) {
272 f.mu.Lock()
273 defer f.mu.Unlock()
274 return f.cfg.BaseURL, f.cfg.InsecureTLS
275}
276
277// SetActiveSession scopes the dashboard view (and Spawn stamping)
278// to a particular host zut session id. Pass empty to clear the
279// scope and revert to "show every agent" (the original behaviour).
280//
281// Existing in-memory agents keep their SessionID; only the filter
282// applied at snapshot time changes. So swapping the active session
283// with /sessions instantly re-narrows the dashboard without
284// touching any agent state.
285func (f *Supervisor) SetActiveSession(id string) {
286 f.mu.Lock()
287 defer f.mu.Unlock()
288 f.activeSession = id
289}
290
291// ActiveSession returns the current scope, mostly for tests and
292// diagnostics. Empty means "no scope; show everything".
293func (f *Supervisor) ActiveSession() string {
294 f.mu.Lock()
295 defer f.mu.Unlock()
296 return f.activeSession
297}
298
299// agentStateDir is the per-agent state directory laid out as:
300//
301// <root>/agents/<id>/
302// events.jsonl durable event log (runner-owned)
303// session.json persistent agent session (child-owned)
304// meta.json static metadata (id, task)
305//
306// The transient unix socket inbox lives in a probed runtime directory,
307// which may be on a different filesystem from this durable state.
308func (f *Supervisor) agentStateDir(id string) string {
309 return filepath.Join(f.cfg.Root, "agents", id)
310}
311
312func (f *Supervisor) uniqueAgentIDLocked(base string) string {
313 id := base
314 for suffix := 2; ; suffix++ {
315 _, inMemory := f.agents[id]
316 _, onDiskErr := os.Stat(f.agentStateDir(id))
317 if !inMemory && onDiskErr != nil {
318 return id
319 }
320 id = fmt.Sprintf("%s-%d", base, suffix)
321 }
322}
323
324// SpawnRequest configures a Spawn. Only Task is required; the rest
325// are optional. Model + Provider, when set, get baked into the
326// child argv as --model / --provider so the agent runs against the
327// chosen model regardless of the parent's current selection. A named
328// Subagent is passed to the child as --subagent, where it resolves the
329// markdown profile and applies its instructions. Reasoning is passed
330// as --reasoning when set. FastMode overrides the host setting when a
331// selected profile declares a fastMode preference.
332type SpawnRequest struct {
333 Task string
334 Model string // optional override; child resolves default if empty
335 Provider string // optional override; usually paired with Model
336 BaseURL string // optional provider endpoint override
337 // InsecureTLS enables the child to skip TLS verification for its
338 // effective custom provider endpoint.
339 InsecureTLS bool
340 Reasoning string // optional reasoning/thinking level override
341 // FastMode is an optional profile override. A non-nil value enables or
342 // disables fast mode for this child; nil inherits Config.FastMode.
343 FastMode *bool
344 Subagent string // optional named markdown profile
345
346 // WebSearchPolicy is an internal child capability override. Generic
347 // requests normally inherit the supervisor's parent decision; named
348 // requests use Allow only when their profile explicitly lists web_search.
349 WebSearchPolicy WebSearchPolicy
350
351 // ParentID is metadata used for per-parent scheduling limits. A
352 // non-empty RequesterAgentID identifies an untrusted child request;
353 // recursive spawning is rejected in v1.
354 ParentID string
355 BatchID string
356 RootSessionID string
357 RequesterAgentID string
358
359 Timeout time.Duration
360 MaxTurns int
361 Tools []string
362 WorkspaceMode WorkspaceMode
363 WorkspaceBase string
364 WorkspaceCapture CaptureMode
365}
366
367// Spawn creates a new Agent for the given task, allocates its on-disk
368// state directory (events log, inbox socket path, session file path), and
369// queues it for the manager-owned scheduler. The returned Agent is pending
370// when the concurrency budget is full and transitions to running when a slot
371// opens. This is the historical signature; callers that want to override the
372// child's model or workspace use SpawnReq instead.
373func (f *Supervisor) Spawn(ctx context.Context, task string) (*Agent, error) {
374 return f.SpawnReq(ctx, SpawnRequest{Task: task})
375}
376
377// SpawnReq is the full-fat variant of Spawn that accepts a SpawnRequest.
378// Existing callers can keep using Spawn; new code that wants to pin the
379// child's model, policy, or workspace uses this.
380func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
381 task := strings.TrimSpace(req.Task)
382 if task == "" {
383 return nil, errors.New("subagents: empty task")
384 }
385 if req.RequesterAgentID != "" {
386 return nil, errors.New("subagents: child-originated spawning is disabled")
387 }
388 if err := f.validateSpawnRequest(req); err != nil {
389 return nil, err
390 }
391 if ctx == nil {
392 ctx = context.Background()
393 }
394
395 now := f.cfg.Now()
396 baseID := newAgentID(task, now)
397
398 f.mu.Lock()
399 id := f.uniqueAgentIDLocked(baseID)
400 dir := f.cfg.RepoRoot
401 fastMode := f.cfg.FastMode
402 configProvider := f.cfg.Provider
403 configBaseURL := f.cfg.BaseURL
404 configInsecureTLS := f.cfg.InsecureTLS
405 sessionID := f.activeSession
406 f.mu.Unlock()
407 fastModeOverridesHost := false
408 if req.FastMode != nil {
409 fastModeOverridesHost = !fastMode && *req.FastMode
410 fastMode = *req.FastMode
411 }
412 childProvider := strings.TrimSpace(req.Provider)
413 inheritProviderSettings := childProvider == "" || configProvider == "" || strings.EqualFold(childProvider, strings.TrimSpace(configProvider))
414 baseURL := strings.TrimSpace(req.BaseURL)
415 if baseURL == "" && inheritProviderSettings {
416 baseURL = strings.TrimSpace(configBaseURL)
417 }
418 insecureTLS := req.InsecureTLS || (inheritProviderSettings && configInsecureTLS)
419
420 stateDir := f.agentStateDir(id)
421 if err := os.MkdirAll(stateDir, 0o700); err != nil {
422 return nil, fmt.Errorf("subagents state dir: %w", err)
423 }
424 lease, err := acquireAgentLease(stateDir)
425 if err != nil {
426 return nil, fmt.Errorf("subagents agent lease: %w", err)
427 }
428 leaseOwned := true
429 defer func() {
430 if leaseOwned {
431 _ = lease.Close()
432 }
433 }()
434 logPath := filepath.Join(stateDir, "events.jsonl")
435 sessionPath := filepath.Join(stateDir, "session.json")
436 // Keep the transient unix socket outside the durable state root.
437 // ZUT_HOME may live on a shared or network filesystem that supports
438 // regular state files but cannot host unix socket nodes.
439 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
440 if err != nil {
441 _ = lease.Close()
442 leaseOwned = false
443 _ = os.RemoveAll(stateDir)
444 return nil, fmt.Errorf("subagents inbox path: %w", err)
445 }
446
447 workspaceMode := req.WorkspaceMode
448 if workspaceMode == "" {
449 workspaceMode = WorkspaceShared
450 }
451 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
452 Mode: workspaceMode,
453 RepositoryRoot: dir,
454 StateDir: stateDir,
455 AgentID: id,
456 Base: req.WorkspaceBase,
457 Capture: req.WorkspaceCapture,
458 AllowedRoots: append([]string(nil), f.cfg.Policy.AllowedRoots...),
459 })
460 if err != nil {
461 _ = lease.Close()
462 leaseOwned = false
463 _ = os.RemoveAll(stateDir)
464 return nil, err
465 }
466 dir = workspace.Dir()
467 maxTurns := req.MaxTurns
468 if maxTurns <= 0 {
469 maxTurns = f.cfg.Policy.MaxTurns
470 }
471 timeout := req.Timeout
472 if timeout <= 0 {
473 timeout = f.cfg.Policy.DefaultTimeout
474 }
475 runCtx, cancel := f.workerContext(timeout)
476 tools := append([]string(nil), req.Tools...)
477 if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
478 tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
479 }
480 webSearchPolicy := f.resolveWebSearchPolicy(req)
481
482 a := &Agent{
483 ID: id,
484 Task: task,
485 OriginalTask: task,
486 Dir: dir,
487 RepositoryRoot: workspace.RepositoryRoot(),
488 WorkspacePath: workspace.Dir(),
489 Started: now,
490 ParentID: strings.TrimSpace(req.ParentID),
491 BatchID: strings.TrimSpace(req.BatchID),
492 RootSessionID: strings.TrimSpace(req.RootSessionID),
493 Model: strings.TrimSpace(req.Model),
494 Provider: strings.TrimSpace(req.Provider),
495 BaseURL: baseURL,
496 InsecureTLS: insecureTLS,
497 Reasoning: strings.TrimSpace(req.Reasoning),
498 FastMode: fastMode,
499 Subagent: strings.TrimSpace(req.Subagent),
500 SessionID: sessionID,
501 WorkspaceMode: workspaceMode,
502 WorkspaceBase: strings.TrimSpace(req.WorkspaceBase),
503 WorkspaceCapture: req.WorkspaceCapture,
504 MaxTurns: maxTurns,
505 Timeout: timeout,
506 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
507 Tools: tools,
508 WebSearchPolicy: webSearchPolicy,
509 InboxPath: inboxPath,
510 EventLogPath: logPath,
511 SessionPath: sessionPath,
512 inbox: NewInbox(inboxPath),
513 status: StatusPending,
514 activity: "queued",
515 transcriptLoaded: true,
516 processState: ProcessPending,
517 turnState: TurnQueued,
518 updatedAt: now,
519 lastActivity: now,
520 stateDir: stateDir,
521 lease: lease,
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 a.fastModeOverridesHost = fastModeOverridesHost
528 if a.RootSessionID == "" {
529 a.RootSessionID = sessionID
530 }
531 a.ctx, a.cancel = runCtx, cancel
532 a.persistFn = f.persistAgent
533 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
534 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
535 a.runner = f.cfg.NewRunner(a)
536 if err := writeAgentMeta(stateDir, a); err != nil {
537 if a.cancel != nil {
538 a.cancel()
539 }
540 _ = workspace.Cleanup(context.Background())
541 _ = lease.Close()
542 leaseOwned = false
543 _ = os.RemoveAll(stateDir)
544 return nil, fmt.Errorf("subagents initial metadata: %w", err)
545 }
546
547 f.mu.Lock()
548 f.agents[id] = a
549 f.order = append(f.order, id)
550 f.queue = append(f.queue, a)
551 f.mu.Unlock()
552 leaseOwned = false
553
554 // Persisted metadata was written before queue admission, so a later
555 // supervisor can reconstruct work that never reached a process.
556 f.armQueueTimeout(a)
557 f.schedule()
558 return a, nil
559}
560
561// SendUserTurn is sugar for the common "send the next user turn"
562// case. It quotes nothing and forwards verbatim; callers are
563// expected to have already trimmed and expanded the text.
564// CancelTurn requests cancellation of the current turn while keeping the
565// worker process alive for a later follow-up.
566func (f *Supervisor) CancelTurn(id string) error {
567 a := f.Get(id)
568 if a == nil {
569 return fmt.Errorf("subagents: no such agent %q", id)
570 }
571 if a.inbox == nil {
572 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
573 }
574 a.setTurnState(TurnCanceling, a.CurrentTurnID())
575 return a.inbox.SendCommand(NewCommand(CommandTurnCancel, a.ID, a.CurrentTurnID(), TurnCancelPayload{Reason: "user"}))
576}
577
578func (f *Supervisor) SendUserTurn(id, text string) error {
579 a := f.Get(id)
580 if a == nil {
581 return fmt.Errorf("subagents: no such agent %q", id)
582 }
583 if a.inbox == nil {
584 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
585 }
586 return a.inbox.SendCommand(NewCommand(CommandTurnStart, a.ID, a.CurrentTurnID(), TurnStartPayload{Prompt: text}))
587}
588
589func (f *Supervisor) run(a *Agent) {
590 a.mu.Lock()
591 killedBeforeStart := a.status == StatusKilled
592 ctxErr := error(nil)
593 if a.ctx != nil {
594 ctxErr = a.ctx.Err()
595 }
596 a.mu.Unlock()
597 var err error
598 if killedBeforeStart || ctxErr != nil || !atomic.CompareAndSwapInt32(&a.launchState, 0, 1) {
599 if ctxErr != nil {
600 err = ctxErr
601 } else {
602 err = context.Canceled
603 }
604 } else {
605 a.setActivity("starting")
606 if a.runner == nil {
607 err = errors.New("subagents: agent has no runner")
608 } else {
609 err = a.runner.Run(a.ctx, agentSink{a: a})
610 }
611 }
612
613 a.mu.Lock()
614 a.finished = f.cfg.Now()
615 switch {
616 case a.status == StatusKilled:
617 // Stop requested a terminal state; preserve it.
618 case errors.Is(err, context.Canceled):
619 a.status = StatusKilled
620 a.activity = "cancelled"
621 a.lastErr = err
622 case errors.Is(err, context.DeadlineExceeded):
623 a.status = StatusFailed
624 a.activity = "timeout"
625 a.lastErr = err
626 case err != nil:
627 a.status = StatusFailed
628 a.activity = "error: " + truncate(err.Error(), 120)
629 a.lastErr = err
630 default:
631 a.status = StatusDone
632 a.activity = "done"
633 }
634 finalStatus := a.status
635 a.mu.Unlock()
636
637 switch finalStatus {
638 case StatusKilled:
639 a.setProcessState(ProcessKilled)
640 a.setTurnState(TurnCanceled, a.CurrentTurnID())
641 case StatusFailed:
642 a.setProcessState(ProcessExited)
643 a.setTurnState(TurnFailed, a.CurrentTurnID())
644 default:
645 a.setProcessState(ProcessExited)
646 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
647 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
648 }
649 }
650
651 // The process slot is no longer occupied once Runner.Run has returned.
652 // Release it before potentially slow artifact capture and workspace
653 // cleanup so queued work can start without waiting on git/filesystem I/O.
654 f.releaseCapacity(a)
655
656 // Capture isolated output and write the structured result before the
657 // temporary workspace is removed. This ordering is the recovery
658 // invariant: a user can always inspect a durable result after failure.
659 f.captureWorkspace(a)
660 f.ensureResult(a, finalStatus, err)
661 if a.cancel != nil {
662 a.cancel()
663 }
664 if a.inbox != nil {
665 _ = a.inbox.Close()
666 }
667 a.setProcessPID(0)
668 f.persistAgent(a)
669 if a.workspaceCleanup != nil {
670 _ = a.workspaceCleanup()
671 }
672 _ = a.releaseLease()
673 a.closeDone()
674}
675
676// List returns a snapshot of every agent in creation order. The
677// returned slice is a copy; callers may iterate without holding the
678// subagent lock. Agent fields are read under their own mutex during
679// formatting in Snapshot.
680func (f *Supervisor) List() []*Agent {
681 f.mu.Lock()
682 defer f.mu.Unlock()
683 out := make([]*Agent, 0, len(f.order))
684 for _, id := range f.order {
685 out = append(out, f.agents[id])
686 }
687 return out
688}
689
690// Get returns the agent with the given (possibly truncated) id, or
691// nil. Matching is prefix-based so the user can type the first few
692// characters of a long id.
693func (f *Supervisor) Get(id string) *Agent {
694 f.mu.Lock()
695 defer f.mu.Unlock()
696 if a, ok := f.agents[id]; ok {
697 return a
698 }
699 // Prefix match.
700 var hits []*Agent
701 for k, a := range f.agents {
702 if strings.HasPrefix(k, id) {
703 hits = append(hits, a)
704 }
705 }
706 if len(hits) == 1 {
707 return hits[0]
708 }
709 return nil
710}
711
712var errDetachedStopTimeout = errors.New("detached worker stop timeout")
713
714const detachedStopPollInterval = 20 * time.Millisecond
715
716// waitForDetachedWorker waits for the worker inbox to disappear, returning a
717// sentinel when the grace period expires. Polling is interruptible so a
718// caller's context is not made to wait for the next fixed sleep interval.
719func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
720 if ctx == nil {
721 ctx = context.Background()
722 }
723 timer := time.NewTimer(timeout)
724 defer timer.Stop()
725 ticker := time.NewTicker(detachedStopPollInterval)
726 defer ticker.Stop()
727 for {
728 if err := ctx.Err(); err != nil {
729 return err
730 }
731 if !inboxLive(path) {
732 return nil
733 }
734 select {
735 case <-ctx.Done():
736 return ctx.Err()
737 case <-timer.C:
738 if !inboxLive(path) {
739 return nil
740 }
741 return errDetachedStopTimeout
742 case <-ticker.C:
743 }
744 }
745}
746
747// Stop cancels the agent's context. The Runner should observe the
748// cancellation and return promptly; the goroutine then finalises the
749// agent in StatusKilled. Also closes the supervisor-side inbox handle
750// so any pending send retries fail fast instead of dialing a
751// socket about to be unlinked. Stop uses a background context; callers
752// that need cancellation while waiting for a detached worker should use
753// StopContext.
754//
755// Stop is a no-op for any agent that's not in a live runnable state
756// — Done / Failed / Killed (already finalised). Reloaded Detached agents
757// are probed first: if their worker survived the supervisor, Stop sends the
758// shutdown command through the durable inbox instead of pretending no
759// process exists.
760func (f *Supervisor) Stop(id string) error {
761 return f.stop(context.Background(), id)
762}
763
764// StopContext is Stop with a context that can interrupt detached-worker
765// shutdown and its graceful and force-stop waits. The context is not held
766// under the supervisor operation lock while those waits run.
767func (f *Supervisor) StopContext(ctx context.Context, id string) error {
768 return f.stop(ctx, id)
769}
770
771func (f *Supervisor) stop(ctx context.Context, id string) error {
772 if ctx == nil {
773 ctx = context.Background()
774 }
775 f.operationMu.Lock()
776 a := f.Get(id)
777 if a == nil {
778 f.operationMu.Unlock()
779 return fmt.Errorf("no such agent %q", id)
780 }
781
782 // Coordinate the status transition with schedule. Holding f.mu while
783 // checking and changing a.status closes the small window where Stop could
784 // observe pending, schedule could promote the same agent to running, and
785 // the pending cleanup below could then finalize an agent that was already
786 // admitted to the runner.
787 f.mu.Lock()
788 a.mu.Lock()
789 status := a.status
790 if status == StatusDone || status == StatusFailed || status == StatusKilled {
791 a.mu.Unlock()
792 f.mu.Unlock()
793 f.operationMu.Unlock()
794 return nil
795 }
796 if status == StatusDetached {
797 path := a.InboxPath
798 inbox := a.inbox
799 pid := a.ProcessPIDValue()
800 a.mu.Unlock()
801 f.mu.Unlock()
802 // The detached worker may take the full grace period to exit. Do not
803 // serialize unrelated supervisor operations for that entire wait.
804 f.operationMu.Unlock()
805 return f.stopDetached(ctx, a, path, inbox, pid)
806 }
807 a.status = StatusKilled
808 a.activity = "canceling"
809 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
810 if status == StatusPending {
811 for i, queued := range f.queue {
812 if queued == a {
813 f.queue = append(f.queue[:i], f.queue[i+1:]...)
814 break
815 }
816 }
817 }
818 a.mu.Unlock()
819 f.mu.Unlock()
820 f.operationMu.Unlock()
821 a.setTurnState(TurnCanceling, a.CurrentTurnID())
822
823 if status == StatusPending {
824 a.setProcessState(ProcessKilled)
825 a.setTurnState(TurnCanceled, a.CurrentTurnID())
826 f.captureWorkspace(a)
827 f.ensureResult(a, StatusKilled, context.Canceled)
828 if a.cancel != nil {
829 a.cancel()
830 }
831 f.persistAgent(a)
832 if a.workspaceCleanup != nil {
833 _ = a.workspaceCleanup()
834 }
835 _ = a.releaseLease()
836 a.closeDone()
837 f.schedule()
838 if err := ctx.Err(); err != nil {
839 return err
840 }
841 return nil
842 }
843
844 // Ask a live worker to shut down cleanly first. If the socket is not
845 // ready, fall back immediately; a forceful context cancellation is
846 // safer than leaving a supervisor entry running forever.
847 if err := ctx.Err(); err != nil {
848 if a.cancel != nil {
849 a.cancel()
850 }
851 return err
852 }
853 graceful := false
854 if a.inbox != nil {
855 err := a.inbox.SendCommand(NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))
856 if err == nil {
857 graceful = true
858 }
859 }
860 if !graceful && a.cancel != nil {
861 a.cancel()
862 }
863 if graceful {
864 grace := f.cfg.Policy.CancelGracePeriod
865 if grace <= 0 {
866 grace = 10 * time.Second
867 }
868 go func() {
869 timer := time.NewTimer(grace)
870 defer timer.Stop()
871 select {
872 case <-a.done:
873 case <-timer.C:
874 if a.cancel != nil {
875 a.cancel()
876 }
877 if a.inbox != nil {
878 _ = a.inbox.Close()
879 }
880 }
881 }()
882 }
883 if err := ctx.Err(); err != nil {
884 if a.cancel != nil {
885 a.cancel()
886 }
887 if a.inbox != nil {
888 _ = a.inbox.Close()
889 }
890 return err
891 }
892 return nil
893}
894
895func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {
896 if inbox == nil {
897 return nil
898 }
899 if err := ctx.Err(); err != nil {
900 return err
901 }
902 if !inboxLive(path) {
903 return nil
904 }
905 if err := ctx.Err(); err != nil {
906 return err
907 }
908 if err := inbox.SendCommand(NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {
909 if ctxErr := ctx.Err(); ctxErr != nil {
910 return ctxErr
911 }
912 return err
913 }
914 a.setActivity("shutdown requested")
915 grace := f.cfg.Policy.CancelGracePeriod
916 if grace <= 0 {
917 grace = 10 * time.Second
918 }
919 waitErr := waitForDetachedWorker(ctx, path, grace)
920 if waitErr != nil && !errors.Is(waitErr, errDetachedStopTimeout) {
921 return waitErr
922 }
923 if waitErr == nil {
924 f.markDetachedKilled(a)
925 return nil
926 }
927 if err := ctx.Err(); err != nil {
928 return err
929 }
930 if pid <= 0 {
931 return fmt.Errorf("agent %s live worker did not stop within %s and has no recorded pid", a.ID, grace)
932 }
933 if killErr := forceKillProcess(pid); killErr != nil {
934 return fmt.Errorf("agent %s force-stop: %w", a.ID, killErr)
935 }
936 if waitErr := waitForDetachedWorker(ctx, path, time.Second); waitErr != nil {
937 if errors.Is(waitErr, errDetachedStopTimeout) {
938 return fmt.Errorf("agent %s live worker did not stop after force-stop", a.ID)
939 }
940 return waitErr
941 }
942 f.markDetachedKilled(a)
943 return nil
944}
945
946func (f *Supervisor) markDetachedKilled(a *Agent) {
947 f.operationMu.Lock()
948 defer f.operationMu.Unlock()
949 a.mu.Lock()
950 a.status = StatusKilled
951 a.activity = "killed"
952 a.mu.Unlock()
953 a.setProcessState(ProcessKilled)
954 a.setTurnState(TurnCanceled, a.CurrentTurnID())
955 a.setProcessPID(0)
956 f.persistAgent(a)
957}
958
959// StopAll cancels every running agent. Used on shutdown.
960func (f *Supervisor) StopAll() {
961 _ = f.stopAll(context.Background())
962}
963
964// StopAllContext is the context-aware shutdown variant. It returns the
965// context error when cancellation interrupts a detached-worker wait.
966func (f *Supervisor) StopAllContext(ctx context.Context) error {
967 return f.stopAll(ctx)
968}
969
970func (f *Supervisor) stopAll(ctx context.Context) error {
971 if ctx == nil {
972 ctx = context.Background()
973 }
974 agents := f.List()
975 var firstErr error
976 for _, a := range agents {
977 if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {
978 firstErr = err
979 }
980 }
981 defer func() {
982 if f.lifetimeCancel != nil {
983 f.lifetimeCancel()
984 }
985 }()
986 for _, a := range agents {
987 select {
988 case <-a.done:
989 case <-ctx.Done():
990 if firstErr == nil {
991 firstErr = ctx.Err()
992 }
993 return firstErr
994 }
995 }
996 return firstErr
997}
998
999// Remove tears down the per-agent state for a terminated agent. It
1000// is an error to remove an agent that's still running; call Stop
1001// first and wait for the status to settle. Detached agents
1002// (reloaded from disk) remove cleanly because they have no live
1003// runner racing for the same files.
1004//
1005// Shared-workspace agents never touch source files during Remove; worktree
1006// agents have already captured their output and removed the isolated checkout
1007// during finalization. Remove then deletes the agent's durable state directory
1008// under <root>/agents/<id>/.
1009func (f *Supervisor) Remove(id string) error {
1010 f.operationMu.Lock()
1011 defer f.operationMu.Unlock()
1012 a := f.Get(id)
1013 if a == nil {
1014 return fmt.Errorf("no such agent %q", id)
1015 }
1016 a.mu.Lock()
1017 st := a.status
1018 inboxPath := a.InboxPath
1019 a.mu.Unlock()
1020 if st == StatusRunning || st == StatusPending {
1021 return fmt.Errorf("agent %s still %s", a.ID, st)
1022 }
1023 if inboxLive(inboxPath) {
1024 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1025 }
1026 select {
1027 case <-a.done:
1028 default:
1029 return fmt.Errorf("agent %s is still finalizing", a.ID)
1030 }
1031 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1032 if err != nil {
1033 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1034 }
1035 _ = lease.Close()
1036 // Best-effort cleanup of the per-agent state directory
1037 // (meta.json, events.jsonl, session.json). Failing here would leave
1038 // the user with no recourse,
1039 // so swallow the error.
1040 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1041 f.mu.Lock()
1042 delete(f.agents, a.ID)
1043 for i, k := range f.order {
1044 if k == a.ID {
1045 f.order = append(f.order[:i], f.order[i+1:]...)
1046 break
1047 }
1048 }
1049 f.mu.Unlock()
1050 return nil
1051}
1052
1053// Snapshot returns a read-only view of one agent. Safe for the TUI
1054// goroutine to call repeatedly; never blocks on the Runner.
1055type AgentSnapshot struct {
1056 ID string
1057 Task string
1058 Dir string
1059 Status Status
1060 ProcessState ProcessState
1061 TurnState TurnState
1062 CurrentTurnID string
1063 Attempt int
1064 Activity string
1065 Started time.Time
1066 Finished time.Time
1067 CreatedAt time.Time
1068 UpdatedAt time.Time
1069 LastActivity time.Time
1070 Err string
1071 Tail string // last few transcript lines, joined with "\n"
1072 Lines []string // full transcript (already capped by Agent.appendTranscript)
1073 LastAssistant string // complete final assistant message, without role formatting
1074 Result *TurnResult
1075 ResultRef string
1076 PatchRef string
1077 ChangedFiles []string
1078 OutputTruncated bool
1079
1080 // Model, Provider, Reasoning, and the provider connection settings
1081 // expose the per-agent configuration captured at Spawn time. Empty
1082 // values inherit the child's defaults. Subagent is the selected named
1083 // markdown profile. The dashboard surfaces these so the user can
1084 // confirm the child configuration.
1085 Model string
1086 Provider string
1087 BaseURL string
1088 InsecureTLS bool
1089 Reasoning string
1090 FastMode bool
1091 Subagent string
1092 WorkspaceMode WorkspaceMode
1093 WorkspacePath string
1094
1095 // Paths to the agent's durable state. Surface them in the
1096 // snapshot so the dashboard /subagents view can
1097 // read events.jsonl or resume the session without going back through the Agent.
1098 InboxPath string
1099 EventLogPath string
1100 SessionPath string
1101}
1102
1103// Snapshot copies the live agent state into a value the caller can
1104// inspect at leisure.
1105func (a *Agent) Snapshot() AgentSnapshot {
1106 a.mu.Lock()
1107 tail := strings.Join(lastN(a.transcript, 6), "\n")
1108 lines := make([]string, len(a.transcript))
1109 copy(lines, a.transcript)
1110 errStr := ""
1111 if a.lastErr != nil {
1112 errStr = a.lastErr.Error()
1113 }
1114 status := a.status
1115 activity := a.activity
1116 lastAssistant := a.lastAssistant
1117 finished := a.finished
1118 a.mu.Unlock()
1119
1120 a.lifecycleMu.Lock()
1121 processState := a.processState
1122 turnState := a.turnState
1123 currentTurnID := a.currentTurnID
1124 attempt := a.Attempt
1125 lastActivity := a.lastActivity
1126 updatedAt := a.updatedAt
1127 result := cloneTurnResult(a.result)
1128 resultRef := a.resultRef
1129 patchRef := a.patchRef
1130 changedFiles := append([]string(nil), a.changedFiles...)
1131 outputTruncated := a.outputTruncated
1132 a.lifecycleMu.Unlock()
1133 return AgentSnapshot{
1134 ID: a.ID, Task: a.Task, Dir: a.Dir,
1135 Status: status, ProcessState: processState, TurnState: turnState,
1136 CurrentTurnID: currentTurnID, Attempt: attempt, Activity: activity,
1137 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1138 Err: errStr, Tail: tail, Lines: lines,
1139 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1140 PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1141 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1142 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1143 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1144 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1145 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1146 }
1147}
1148
1149// SnapshotAll returns snapshots of every agent in creation order,
1150// scoped to the active session when one is set.
1151//
1152// Scoping rules:
1153// - activeSession == "": no filter, every agent is returned
1154// (historical behaviour; used by tests and scripted callers).
1155// - activeSession != "": include live agents whose SessionID matches
1156// activeSession, plus unscoped agents. Detached and terminal
1157// agents from another host session remain visible so restart/recovery
1158// does not hide durable work.
1159func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1160 agents := f.List()
1161 f.mu.Lock()
1162 active := f.activeSession
1163 f.mu.Unlock()
1164
1165 out := make([]AgentSnapshot, 0, len(agents))
1166 for _, a := range agents {
1167 if active != "" && a.SessionID != "" && a.SessionID != active {
1168 // Keep detached/terminal entries visible after a normal host
1169 // restart so durable recovery remains discoverable. Hide only
1170 // live work belonging to another in-process host session.
1171 status := a.Status()
1172 if status == StatusRunning || status == StatusPending {
1173 continue
1174 }
1175 }
1176 out = append(out, a.Snapshot())
1177 }
1178 // Sort by start time for a stable, deterministic listing.
1179 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1180 return out
1181}
1182
1183// agentSink is the Sink the Supervisor hands to each Runner.
1184type agentSink struct{ a *Agent }
1185
1186func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1187func (s agentSink) Transcript(chunk string) {
1188 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") {
1189 s.a.appendTranscript(chunk)
1190 return
1191 }
1192 s.a.appendAssistantMessage(chunk)
1193}
1194func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1195func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1196
1197func truncate(s string, n int) string {
1198 runes := []rune(s)
1199 if len(runes) <= n {
1200 return s
1201 }
1202 if n <= 3 {
1203 return strings.Repeat(".", n)
1204 }
1205 return string(runes[:n-3]) + "..."
1206}
1207
1208func lastN(lines []string, n int) []string {
1209 if len(lines) <= n {
1210 return lines
1211 }
1212 return lines[len(lines)-n:]
1213}