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