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