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 restricts 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 restriction. A non-nil false value
321 // disables fast mode for this child; true still cannot enable fast mode
322 // when Config.FastMode is false. Nil inherits Config.FastMode.
323 FastMode *bool
324 Subagent string // optional named markdown profile
325
326 // ParentID is metadata used for per-parent scheduling limits. A
327 // non-empty RequesterAgentID identifies an untrusted child request;
328 // recursive spawning is rejected in v1.
329 ParentID string
330 BatchID string
331 RootSessionID string
332 RequesterAgentID string
333
334 Timeout time.Duration
335 MaxTurns int
336 Tools []string
337 WorkspaceMode WorkspaceMode
338 WorkspaceBase string
339 WorkspaceCapture CaptureMode
340}
341
342// Spawn creates a new Agent for the given task, allocates its on-disk
343// state directory (events log, inbox socket path, session file path), and
344// queues it for the manager-owned scheduler. The returned Agent is pending
345// when the concurrency budget is full and transitions to running when a slot
346// opens. This is the historical signature; callers that want to override the
347// child's model or workspace use SpawnReq instead.
348func (f *Supervisor) Spawn(ctx context.Context, task string) (*Agent, error) {
349 return f.SpawnReq(ctx, SpawnRequest{Task: task})
350}
351
352// SpawnReq is the full-fat variant of Spawn that accepts a SpawnRequest.
353// Existing callers can keep using Spawn; new code that wants to pin the
354// child's model, policy, or workspace uses this.
355func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
356 task := strings.TrimSpace(req.Task)
357 if task == "" {
358 return nil, errors.New("subagents: empty task")
359 }
360 if req.RequesterAgentID != "" {
361 return nil, errors.New("subagents: child-originated spawning is disabled")
362 }
363 if err := f.validateSpawnRequest(req); err != nil {
364 return nil, err
365 }
366 if ctx == nil {
367 ctx = context.Background()
368 }
369
370 now := f.cfg.Now()
371 baseID := newAgentID(task, now)
372
373 // Reserve a spawn budget before touching disk. This makes the total
374 // budget race-free across slash commands, tools, and batch callers.
375 f.mu.Lock()
376 id := f.uniqueAgentIDLocked(baseID)
377 if f.totalSpawned >= f.cfg.Policy.MaxTotalSpawned {
378 f.mu.Unlock()
379 return nil, fmt.Errorf("subagents: total spawn budget exhausted (%d)", f.cfg.Policy.MaxTotalSpawned)
380 }
381 f.totalSpawned++
382 dir := f.cfg.RepoRoot
383 fastMode := f.cfg.FastMode
384 configProvider := f.cfg.Provider
385 configBaseURL := f.cfg.BaseURL
386 configInsecureTLS := f.cfg.InsecureTLS
387 sessionID := f.activeSession
388 f.mu.Unlock()
389 if req.FastMode != nil {
390 fastMode = fastMode && *req.FastMode
391 }
392 childProvider := strings.TrimSpace(req.Provider)
393 inheritProviderSettings := childProvider == "" || configProvider == "" || strings.EqualFold(childProvider, strings.TrimSpace(configProvider))
394 baseURL := strings.TrimSpace(req.BaseURL)
395 if baseURL == "" && inheritProviderSettings {
396 baseURL = strings.TrimSpace(configBaseURL)
397 }
398 insecureTLS := req.InsecureTLS || (inheritProviderSettings && configInsecureTLS)
399 reserved := true
400 defer func() {
401 if reserved {
402 f.mu.Lock()
403 f.totalSpawned--
404 f.mu.Unlock()
405 }
406 }()
407
408 stateDir := f.agentStateDir(id)
409 if err := os.MkdirAll(stateDir, 0o700); err != nil {
410 return nil, fmt.Errorf("subagents state dir: %w", err)
411 }
412 lease, err := acquireAgentLease(stateDir)
413 if err != nil {
414 return nil, fmt.Errorf("subagents agent lease: %w", err)
415 }
416 leaseOwned := true
417 defer func() {
418 if leaseOwned {
419 _ = lease.Close()
420 }
421 }()
422 logPath := filepath.Join(stateDir, "events.jsonl")
423 sessionPath := filepath.Join(stateDir, "session.json")
424 // Keep the transient unix socket outside the durable state root.
425 // ZUT_HOME may live on a shared or network filesystem that supports
426 // regular state files but cannot host unix socket nodes.
427 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
428 if err != nil {
429 _ = lease.Close()
430 leaseOwned = false
431 _ = os.RemoveAll(stateDir)
432 return nil, fmt.Errorf("subagents inbox path: %w", err)
433 }
434
435 workspaceMode := req.WorkspaceMode
436 if workspaceMode == "" {
437 workspaceMode = WorkspaceShared
438 }
439 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
440 Mode: workspaceMode,
441 RepositoryRoot: dir,
442 StateDir: stateDir,
443 AgentID: id,
444 Base: req.WorkspaceBase,
445 Capture: req.WorkspaceCapture,
446 AllowedRoots: append([]string(nil), f.cfg.Policy.AllowedRoots...),
447 })
448 if err != nil {
449 _ = lease.Close()
450 leaseOwned = false
451 _ = os.RemoveAll(stateDir)
452 return nil, err
453 }
454 dir = workspace.Dir()
455 maxTurns := req.MaxTurns
456 if maxTurns <= 0 {
457 maxTurns = f.cfg.Policy.MaxTurns
458 }
459 timeout := req.Timeout
460 if timeout <= 0 {
461 timeout = f.cfg.Policy.DefaultTimeout
462 }
463 runCtx, cancel := f.workerContext(timeout)
464 tools := append([]string(nil), req.Tools...)
465 if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
466 tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
467 }
468
469 a := &Agent{
470 ID: id,
471 Task: task,
472 OriginalTask: task,
473 Dir: dir,
474 RepositoryRoot: workspace.RepositoryRoot(),
475 WorkspacePath: workspace.Dir(),
476 Started: now,
477 ParentID: strings.TrimSpace(req.ParentID),
478 BatchID: strings.TrimSpace(req.BatchID),
479 RootSessionID: strings.TrimSpace(req.RootSessionID),
480 Model: strings.TrimSpace(req.Model),
481 Provider: strings.TrimSpace(req.Provider),
482 BaseURL: baseURL,
483 InsecureTLS: insecureTLS,
484 Reasoning: strings.TrimSpace(req.Reasoning),
485 FastMode: fastMode,
486 Subagent: strings.TrimSpace(req.Subagent),
487 SessionID: sessionID,
488 WorkspaceMode: workspaceMode,
489 WorkspaceBase: strings.TrimSpace(req.WorkspaceBase),
490 WorkspaceCapture: req.WorkspaceCapture,
491 MaxTurns: maxTurns,
492 Timeout: timeout,
493 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
494 Tools: tools,
495 InboxPath: inboxPath,
496 EventLogPath: logPath,
497 SessionPath: sessionPath,
498 inbox: NewInbox(inboxPath),
499 status: StatusPending,
500 activity: "queued",
501 processState: ProcessPending,
502 turnState: TurnQueued,
503 updatedAt: now,
504 lastActivity: now,
505 stateDir: stateDir,
506 lease: lease,
507 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
508 maxOutputLines: f.cfg.Policy.MaxOutputLines,
509 done: make(chan struct{}),
510 turnResults: make(chan *TurnResult, 16),
511 }
512 if a.RootSessionID == "" {
513 a.RootSessionID = sessionID
514 }
515 a.ctx, a.cancel = runCtx, cancel
516 a.persistFn = f.persistAgent
517 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
518 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
519 a.runner = f.cfg.NewRunner(a)
520 if err := writeAgentMeta(stateDir, a); err != nil {
521 if a.cancel != nil {
522 a.cancel()
523 }
524 _ = workspace.Cleanup(context.Background())
525 _ = lease.Close()
526 leaseOwned = false
527 _ = os.RemoveAll(stateDir)
528 return nil, fmt.Errorf("subagents initial metadata: %w", err)
529 }
530
531 f.mu.Lock()
532 f.agents[id] = a
533 f.order = append(f.order, id)
534 f.queue = append(f.queue, a)
535 f.mu.Unlock()
536 leaseOwned = false
537
538 // Persisted metadata was written before queue admission, so a later
539 // supervisor can reconstruct work that never reached a process.
540 f.armQueueTimeout(a)
541 f.schedule()
542 reserved = false
543 return a, nil
544}
545
546// SendUserTurn is sugar for the common "send the next user turn"
547// case. It quotes nothing and forwards verbatim; callers are
548// expected to have already trimmed and expanded the text.
549// CancelTurn requests cancellation of the current turn while keeping the
550// worker process alive for a later follow-up.
551func (f *Supervisor) CancelTurn(id string) error {
552 a := f.Get(id)
553 if a == nil {
554 return fmt.Errorf("subagents: no such agent %q", id)
555 }
556 if a.inbox == nil {
557 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
558 }
559 a.setTurnState(TurnCanceling, a.CurrentTurnID())
560 return a.inbox.SendCommand(NewCommand(CommandTurnCancel, a.ID, a.CurrentTurnID(), TurnCancelPayload{Reason: "user"}))
561}
562
563func (f *Supervisor) SendUserTurn(id, text string) error {
564 a := f.Get(id)
565 if a == nil {
566 return fmt.Errorf("subagents: no such agent %q", id)
567 }
568 if a.inbox == nil {
569 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
570 }
571 return a.inbox.SendCommand(NewCommand(CommandTurnStart, a.ID, a.CurrentTurnID(), TurnStartPayload{Prompt: text}))
572}
573
574func (f *Supervisor) run(a *Agent) {
575 a.mu.Lock()
576 killedBeforeStart := a.status == StatusKilled
577 ctxErr := error(nil)
578 if a.ctx != nil {
579 ctxErr = a.ctx.Err()
580 }
581 a.mu.Unlock()
582 var err error
583 if killedBeforeStart || ctxErr != nil || !atomic.CompareAndSwapInt32(&a.launchState, 0, 1) {
584 if ctxErr != nil {
585 err = ctxErr
586 } else {
587 err = context.Canceled
588 }
589 } else {
590 a.setActivity("starting")
591 if a.runner == nil {
592 err = errors.New("subagents: agent has no runner")
593 } else {
594 err = a.runner.Run(a.ctx, agentSink{a: a})
595 }
596 }
597
598 a.mu.Lock()
599 a.finished = f.cfg.Now()
600 switch {
601 case a.status == StatusKilled:
602 // Stop requested a terminal state; preserve it.
603 case errors.Is(err, context.Canceled):
604 a.status = StatusKilled
605 a.activity = "cancelled"
606 a.lastErr = err
607 case errors.Is(err, context.DeadlineExceeded):
608 a.status = StatusFailed
609 a.activity = "timeout"
610 a.lastErr = err
611 case err != nil:
612 a.status = StatusFailed
613 a.activity = "error: " + truncate(err.Error(), 120)
614 a.lastErr = err
615 default:
616 a.status = StatusDone
617 a.activity = "done"
618 }
619 finalStatus := a.status
620 a.mu.Unlock()
621
622 switch finalStatus {
623 case StatusKilled:
624 a.setProcessState(ProcessKilled)
625 a.setTurnState(TurnCanceled, a.CurrentTurnID())
626 case StatusFailed:
627 a.setProcessState(ProcessExited)
628 a.setTurnState(TurnFailed, a.CurrentTurnID())
629 default:
630 a.setProcessState(ProcessExited)
631 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
632 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
633 }
634 }
635
636 // The process slot is no longer occupied once Runner.Run has returned.
637 // Release it before potentially slow artifact capture and workspace
638 // cleanup so queued work can start without waiting on git/filesystem I/O.
639 f.releaseCapacity(a)
640
641 // Capture isolated output and write the structured result before the
642 // temporary workspace is removed. This ordering is the recovery
643 // invariant: a user can always inspect a durable result after failure.
644 f.captureWorkspace(a)
645 f.ensureResult(a, finalStatus, err)
646 if a.cancel != nil {
647 a.cancel()
648 }
649 if a.inbox != nil {
650 _ = a.inbox.Close()
651 }
652 a.setProcessPID(0)
653 f.persistAgent(a)
654 if a.workspaceCleanup != nil {
655 _ = a.workspaceCleanup()
656 }
657 _ = a.releaseLease()
658 a.closeDone()
659}
660
661// List returns a snapshot of every agent in creation order. The
662// returned slice is a copy; callers may iterate without holding the
663// subagent lock. Agent fields are read under their own mutex during
664// formatting in Snapshot.
665func (f *Supervisor) List() []*Agent {
666 f.mu.Lock()
667 defer f.mu.Unlock()
668 out := make([]*Agent, 0, len(f.order))
669 for _, id := range f.order {
670 out = append(out, f.agents[id])
671 }
672 return out
673}
674
675// Get returns the agent with the given (possibly truncated) id, or
676// nil. Matching is prefix-based so the user can type the first few
677// characters of a long id.
678func (f *Supervisor) Get(id string) *Agent {
679 f.mu.Lock()
680 defer f.mu.Unlock()
681 if a, ok := f.agents[id]; ok {
682 return a
683 }
684 // Prefix match.
685 var hits []*Agent
686 for k, a := range f.agents {
687 if strings.HasPrefix(k, id) {
688 hits = append(hits, a)
689 }
690 }
691 if len(hits) == 1 {
692 return hits[0]
693 }
694 return nil
695}
696
697var errDetachedStopTimeout = errors.New("detached worker stop timeout")
698
699const detachedStopPollInterval = 20 * time.Millisecond
700
701// waitForDetachedWorker waits for the worker inbox to disappear, returning a
702// sentinel when the grace period expires. Polling is interruptible so a
703// caller's context is not made to wait for the next fixed sleep interval.
704func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
705 if ctx == nil {
706 ctx = context.Background()
707 }
708 timer := time.NewTimer(timeout)
709 defer timer.Stop()
710 ticker := time.NewTicker(detachedStopPollInterval)
711 defer ticker.Stop()
712 for {
713 if err := ctx.Err(); err != nil {
714 return err
715 }
716 if !inboxLive(path) {
717 return nil
718 }
719 select {
720 case <-ctx.Done():
721 return ctx.Err()
722 case <-timer.C:
723 if !inboxLive(path) {
724 return nil
725 }
726 return errDetachedStopTimeout
727 case <-ticker.C:
728 }
729 }
730}
731
732// Stop cancels the agent's context. The Runner should observe the
733// cancellation and return promptly; the goroutine then finalises the
734// agent in StatusKilled. Also closes the supervisor-side inbox handle
735// so any pending send retries fail fast instead of dialing a
736// socket about to be unlinked. Stop uses a background context; callers
737// that need cancellation while waiting for a detached worker should use
738// StopContext.
739//
740// Stop is a no-op for any agent that's not in a live runnable state
741// — Done / Failed / Killed (already finalised). Reloaded Detached agents
742// are probed first: if their worker survived the supervisor, Stop sends the
743// shutdown command through the durable inbox instead of pretending no
744// process exists.
745func (f *Supervisor) Stop(id string) error {
746 return f.stop(context.Background(), id)
747}
748
749// StopContext is Stop with a context that can interrupt detached-worker
750// shutdown and its graceful and force-stop waits. The context is not held
751// under the supervisor operation lock while those waits run.
752func (f *Supervisor) StopContext(ctx context.Context, id string) error {
753 return f.stop(ctx, id)
754}
755
756func (f *Supervisor) stop(ctx context.Context, id string) error {
757 if ctx == nil {
758 ctx = context.Background()
759 }
760 f.operationMu.Lock()
761 a := f.Get(id)
762 if a == nil {
763 f.operationMu.Unlock()
764 return fmt.Errorf("no such agent %q", id)
765 }
766
767 // Coordinate the status transition with schedule. Holding f.mu while
768 // checking and changing a.status closes the small window where Stop could
769 // observe pending, schedule could promote the same agent to running, and
770 // the pending cleanup below could then finalize an agent that was already
771 // admitted to the runner.
772 f.mu.Lock()
773 a.mu.Lock()
774 status := a.status
775 if status == StatusDone || status == StatusFailed || status == StatusKilled {
776 a.mu.Unlock()
777 f.mu.Unlock()
778 f.operationMu.Unlock()
779 return nil
780 }
781 if status == StatusDetached {
782 path := a.InboxPath
783 inbox := a.inbox
784 pid := a.ProcessPIDValue()
785 a.mu.Unlock()
786 f.mu.Unlock()
787 // The detached worker may take the full grace period to exit. Do not
788 // serialize unrelated supervisor operations for that entire wait.
789 f.operationMu.Unlock()
790 return f.stopDetached(ctx, a, path, inbox, pid)
791 }
792 a.status = StatusKilled
793 a.activity = "canceling"
794 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
795 if status == StatusPending {
796 for i, queued := range f.queue {
797 if queued == a {
798 f.queue = append(f.queue[:i], f.queue[i+1:]...)
799 break
800 }
801 }
802 }
803 a.mu.Unlock()
804 f.mu.Unlock()
805 f.operationMu.Unlock()
806 a.setTurnState(TurnCanceling, a.CurrentTurnID())
807
808 if status == StatusPending {
809 a.setProcessState(ProcessKilled)
810 a.setTurnState(TurnCanceled, a.CurrentTurnID())
811 f.captureWorkspace(a)
812 f.ensureResult(a, StatusKilled, context.Canceled)
813 if a.cancel != nil {
814 a.cancel()
815 }
816 f.persistAgent(a)
817 if a.workspaceCleanup != nil {
818 _ = a.workspaceCleanup()
819 }
820 _ = a.releaseLease()
821 a.closeDone()
822 f.schedule()
823 if err := ctx.Err(); err != nil {
824 return err
825 }
826 return nil
827 }
828
829 // Ask a live worker to shut down cleanly first. If the socket is not
830 // ready, fall back immediately; a forceful context cancellation is
831 // safer than leaving a supervisor entry running forever.
832 if err := ctx.Err(); err != nil {
833 if a.cancel != nil {
834 a.cancel()
835 }
836 return err
837 }
838 graceful := false
839 if a.inbox != nil {
840 err := a.inbox.SendCommand(NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))
841 if err == nil {
842 graceful = true
843 }
844 }
845 if !graceful && a.cancel != nil {
846 a.cancel()
847 }
848 if graceful {
849 grace := f.cfg.Policy.CancelGracePeriod
850 if grace <= 0 {
851 grace = 10 * time.Second
852 }
853 go func() {
854 timer := time.NewTimer(grace)
855 defer timer.Stop()
856 select {
857 case <-a.done:
858 case <-timer.C:
859 if a.cancel != nil {
860 a.cancel()
861 }
862 if a.inbox != nil {
863 _ = a.inbox.Close()
864 }
865 }
866 }()
867 }
868 if err := ctx.Err(); err != nil {
869 if a.cancel != nil {
870 a.cancel()
871 }
872 if a.inbox != nil {
873 _ = a.inbox.Close()
874 }
875 return err
876 }
877 return nil
878}
879
880func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {
881 if inbox == nil {
882 return nil
883 }
884 if err := ctx.Err(); err != nil {
885 return err
886 }
887 if !inboxLive(path) {
888 return nil
889 }
890 if err := ctx.Err(); err != nil {
891 return err
892 }
893 if err := inbox.SendCommand(NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {
894 if ctxErr := ctx.Err(); ctxErr != nil {
895 return ctxErr
896 }
897 return err
898 }
899 a.setActivity("shutdown requested")
900 grace := f.cfg.Policy.CancelGracePeriod
901 if grace <= 0 {
902 grace = 10 * time.Second
903 }
904 waitErr := waitForDetachedWorker(ctx, path, grace)
905 if waitErr != nil && !errors.Is(waitErr, errDetachedStopTimeout) {
906 return waitErr
907 }
908 if waitErr == nil {
909 f.markDetachedKilled(a)
910 return nil
911 }
912 if err := ctx.Err(); err != nil {
913 return err
914 }
915 if pid <= 0 {
916 return fmt.Errorf("agent %s live worker did not stop within %s and has no recorded pid", a.ID, grace)
917 }
918 if killErr := forceKillProcess(pid); killErr != nil {
919 return fmt.Errorf("agent %s force-stop: %w", a.ID, killErr)
920 }
921 if waitErr := waitForDetachedWorker(ctx, path, time.Second); waitErr != nil {
922 if errors.Is(waitErr, errDetachedStopTimeout) {
923 return fmt.Errorf("agent %s live worker did not stop after force-stop", a.ID)
924 }
925 return waitErr
926 }
927 f.markDetachedKilled(a)
928 return nil
929}
930
931func (f *Supervisor) markDetachedKilled(a *Agent) {
932 f.operationMu.Lock()
933 defer f.operationMu.Unlock()
934 a.mu.Lock()
935 a.status = StatusKilled
936 a.activity = "killed"
937 a.mu.Unlock()
938 a.setProcessState(ProcessKilled)
939 a.setTurnState(TurnCanceled, a.CurrentTurnID())
940 a.setProcessPID(0)
941 f.persistAgent(a)
942}
943
944// StopAll cancels every running agent. Used on shutdown.
945func (f *Supervisor) StopAll() {
946 _ = f.stopAll(context.Background())
947}
948
949// StopAllContext is the context-aware shutdown variant. It returns the
950// context error when cancellation interrupts a detached-worker wait.
951func (f *Supervisor) StopAllContext(ctx context.Context) error {
952 return f.stopAll(ctx)
953}
954
955func (f *Supervisor) stopAll(ctx context.Context) error {
956 if ctx == nil {
957 ctx = context.Background()
958 }
959 agents := f.List()
960 var firstErr error
961 for _, a := range agents {
962 if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {
963 firstErr = err
964 }
965 }
966 defer func() {
967 if f.lifetimeCancel != nil {
968 f.lifetimeCancel()
969 }
970 }()
971 for _, a := range agents {
972 select {
973 case <-a.done:
974 case <-ctx.Done():
975 if firstErr == nil {
976 firstErr = ctx.Err()
977 }
978 return firstErr
979 }
980 }
981 return firstErr
982}
983
984// Remove tears down the per-agent state for a terminated agent. It
985// is an error to remove an agent that's still running; call Stop
986// first and wait for the status to settle. Detached agents
987// (reloaded from disk) remove cleanly because they have no live
988// runner racing for the same files.
989//
990// Shared-workspace agents never touch source files during Remove; worktree
991// agents have already captured their output and removed the isolated checkout
992// during finalization. Remove then deletes the agent's durable state directory
993// under <root>/agents/<id>/.
994func (f *Supervisor) Remove(id string) error {
995 f.operationMu.Lock()
996 defer f.operationMu.Unlock()
997 a := f.Get(id)
998 if a == nil {
999 return fmt.Errorf("no such agent %q", id)
1000 }
1001 a.mu.Lock()
1002 st := a.status
1003 inboxPath := a.InboxPath
1004 a.mu.Unlock()
1005 if st == StatusRunning || st == StatusPending {
1006 return fmt.Errorf("agent %s still %s", a.ID, st)
1007 }
1008 if inboxLive(inboxPath) {
1009 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1010 }
1011 select {
1012 case <-a.done:
1013 default:
1014 return fmt.Errorf("agent %s is still finalizing", a.ID)
1015 }
1016 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1017 if err != nil {
1018 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1019 }
1020 _ = lease.Close()
1021 // Best-effort cleanup of the per-agent state directory
1022 // (meta.json, events.jsonl, session.json). Failing here would leave
1023 // the user with no recourse,
1024 // so swallow the error.
1025 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1026 f.mu.Lock()
1027 delete(f.agents, a.ID)
1028 for i, k := range f.order {
1029 if k == a.ID {
1030 f.order = append(f.order[:i], f.order[i+1:]...)
1031 break
1032 }
1033 }
1034 f.mu.Unlock()
1035 return nil
1036}
1037
1038// Snapshot returns a read-only view of one agent. Safe for the TUI
1039// goroutine to call repeatedly; never blocks on the Runner.
1040type AgentSnapshot struct {
1041 ID string
1042 Task string
1043 Dir string
1044 Status Status
1045 ProcessState ProcessState
1046 TurnState TurnState
1047 CurrentTurnID string
1048 Attempt int
1049 Activity string
1050 Started time.Time
1051 Finished time.Time
1052 CreatedAt time.Time
1053 UpdatedAt time.Time
1054 LastActivity time.Time
1055 Err string
1056 Tail string // last few transcript lines, joined with "\n"
1057 Lines []string // full transcript (already capped by Agent.appendTranscript)
1058 LastAssistant string // complete final assistant message, without role formatting
1059 Result *TurnResult
1060 ResultRef string
1061 PatchRef string
1062 ChangedFiles []string
1063 OutputTruncated bool
1064
1065 // Model, Provider, Reasoning, and the provider connection settings
1066 // expose the per-agent configuration captured at Spawn time. Empty
1067 // values inherit the child's defaults. Subagent is the selected named
1068 // markdown profile. The dashboard surfaces these so the user can
1069 // confirm the child configuration.
1070 Model string
1071 Provider string
1072 BaseURL string
1073 InsecureTLS bool
1074 Reasoning string
1075 FastMode bool
1076 Subagent string
1077 WorkspaceMode WorkspaceMode
1078 WorkspacePath string
1079
1080 // Paths to the agent's durable state. Surface them in the
1081 // snapshot so the dashboard /subagents view can
1082 // read events.jsonl or resume the session without going back through the Agent.
1083 InboxPath string
1084 EventLogPath string
1085 SessionPath string
1086}
1087
1088// Snapshot copies the live agent state into a value the caller can
1089// inspect at leisure.
1090func (a *Agent) Snapshot() AgentSnapshot {
1091 a.mu.Lock()
1092 tail := strings.Join(lastN(a.transcript, 6), "\n")
1093 lines := make([]string, len(a.transcript))
1094 copy(lines, a.transcript)
1095 errStr := ""
1096 if a.lastErr != nil {
1097 errStr = a.lastErr.Error()
1098 }
1099 status := a.status
1100 activity := a.activity
1101 lastAssistant := a.lastAssistant
1102 finished := a.finished
1103 a.mu.Unlock()
1104
1105 a.lifecycleMu.Lock()
1106 processState := a.processState
1107 turnState := a.turnState
1108 currentTurnID := a.currentTurnID
1109 attempt := a.Attempt
1110 lastActivity := a.lastActivity
1111 updatedAt := a.updatedAt
1112 result := cloneTurnResult(a.result)
1113 resultRef := a.resultRef
1114 patchRef := a.patchRef
1115 changedFiles := append([]string(nil), a.changedFiles...)
1116 outputTruncated := a.outputTruncated
1117 a.lifecycleMu.Unlock()
1118 return AgentSnapshot{
1119 ID: a.ID, Task: a.Task, Dir: a.Dir,
1120 Status: status, ProcessState: processState, TurnState: turnState,
1121 CurrentTurnID: currentTurnID, Attempt: attempt, Activity: activity,
1122 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1123 Err: errStr, Tail: tail, Lines: lines,
1124 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1125 PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1126 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1127 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1128 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1129 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1130 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1131 }
1132}
1133
1134// SnapshotAll returns snapshots of every agent in creation order,
1135// scoped to the active session when one is set.
1136//
1137// Scoping rules:
1138// - activeSession == "": no filter, every agent is returned
1139// (historical behaviour; used by tests and scripted callers).
1140// - activeSession != "": include live agents whose SessionID matches
1141// activeSession, plus unscoped agents. Detached and terminal
1142// agents from another host session remain visible so restart/recovery
1143// does not hide durable work.
1144func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1145 agents := f.List()
1146 f.mu.Lock()
1147 active := f.activeSession
1148 f.mu.Unlock()
1149
1150 out := make([]AgentSnapshot, 0, len(agents))
1151 for _, a := range agents {
1152 if active != "" && a.SessionID != "" && a.SessionID != active {
1153 // Keep detached/terminal entries visible after a normal host
1154 // restart so durable recovery remains discoverable. Hide only
1155 // live work belonging to another in-process host session.
1156 status := a.Status()
1157 if status == StatusRunning || status == StatusPending {
1158 continue
1159 }
1160 }
1161 out = append(out, a.Snapshot())
1162 }
1163 // Sort by start time for a stable, deterministic listing.
1164 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1165 return out
1166}
1167
1168// agentSink is the Sink the Supervisor hands to each Runner.
1169type agentSink struct{ a *Agent }
1170
1171func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1172func (s agentSink) Transcript(chunk string) {
1173 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") {
1174 s.a.appendTranscript(chunk)
1175 return
1176 }
1177 s.a.appendAssistantMessage(chunk)
1178}
1179func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1180func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1181
1182func truncate(s string, n int) string {
1183 runes := []rune(s)
1184 if len(runes) <= n {
1185 return s
1186 }
1187 if n <= 3 {
1188 return strings.Repeat(".", n)
1189 }
1190 return string(runes[:n-3]) + "..."
1191}
1192
1193func lastN(lines []string, n int) []string {
1194 if len(lines) <= n {
1195 return lines
1196 }
1197 return lines[len(lines)-n:]
1198}