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{
159 agent: a,
160 resolveCredential: cfg.ResolveCredential,
161 GracePeriod: cfg.Policy.CancelGracePeriod,
162 }
163 }
164 }
165 cfg.Policy.normalize()
166 lifetimeParent := cfg.Context
167 if lifetimeParent == nil {
168 lifetimeParent = context.Background()
169 }
170 lifetimeCtx, lifetimeCancel := context.WithCancel(lifetimeParent)
171 return &Supervisor{
172 cfg: cfg,
173 lifetimeCtx: lifetimeCtx,
174 lifetimeCancel: lifetimeCancel,
175 agents: map[string]*Agent{},
176 activeByParent: map[string]int{},
177 activeByBatch: map[string]int{},
178 batches: map[string]*Batch{},
179 }
180}
181
182// workerContext derives a child worker context from the supervisor lifetime.
183// Turn timeouts are enforced by the worker while a delegated turn is active;
184// the process remains cancellable while idle so it can accept follow-ups.
185func (f *Supervisor) workerContext() (context.Context, context.CancelFunc) {
186 base := f.lifetimeCtx
187 if base == nil {
188 base = context.Background()
189 }
190 return context.WithCancel(base)
191}
192
193// SetRepoRoot updates the shared working directory for subsequently
194// spawned agents. Existing agents keep the directory they were started
195// with; callers use this when the host session changes cwd.
196func (f *Supervisor) SetRepoRoot(root string) {
197 f.mu.Lock()
198 defer f.mu.Unlock()
199 f.cfg.RepoRoot = root
200}
201
202// RepoRoot returns the working directory used by subsequently spawned
203// agents.
204func (f *Supervisor) RepoRoot() string {
205 f.mu.Lock()
206 defer f.mu.Unlock()
207 return f.cfg.RepoRoot
208}
209
210// MaxTurns returns the maximum prompt-level turns allowed for subsequently
211// spawned agents.
212func (f *Supervisor) MaxTurns() int {
213 f.mu.Lock()
214 defer f.mu.Unlock()
215 return f.cfg.Policy.MaxTurns
216}
217
218// SetFastMode updates the fast-mode setting for subsequently spawned
219// children. Existing agents keep the setting captured at their spawn
220// boundary so resumes remain deterministic.
221func (f *Supervisor) SetFastMode(enabled bool) {
222 f.mu.Lock()
223 f.cfg.FastMode = enabled
224 f.mu.Unlock()
225}
226
227// FastMode reports the setting used for subsequently spawned children.
228func (f *Supervisor) FastMode() bool {
229 f.mu.Lock()
230 defer f.mu.Unlock()
231 return f.cfg.FastMode
232}
233
234// SetWebSearchPolicy updates the parent capability used for subsequently
235// spawned generic children. Existing agents retain their persisted decision.
236func (f *Supervisor) SetWebSearchPolicy(policy WebSearchPolicy) {
237 f.mu.Lock()
238 f.cfg.WebSearchPolicy = policy
239 f.mu.Unlock()
240}
241
242// WebSearchPolicy reports the parent capability used for new generic
243// children.
244func (f *Supervisor) WebSearchPolicy() WebSearchPolicy {
245 f.mu.Lock()
246 defer f.mu.Unlock()
247 return f.cfg.WebSearchPolicy
248}
249
250// SetProvider updates the provider used by subsequently spawned children.
251// Existing agents keep their persisted provider selection.
252func (f *Supervisor) SetProvider(provider string) {
253 f.mu.Lock()
254 f.cfg.Provider = strings.TrimSpace(provider)
255 f.mu.Unlock()
256}
257
258// SetProviderSettings updates the effective connection settings used by
259// subsequently spawned children. Existing agents keep their persisted values
260// so a model/provider switch cannot silently change a running or resumed
261// worker's endpoint.
262func (f *Supervisor) SetProviderSettings(baseURL string, insecureTLS bool) {
263 f.mu.Lock()
264 f.cfg.BaseURL = strings.TrimSpace(baseURL)
265 f.cfg.InsecureTLS = insecureTLS
266 f.mu.Unlock()
267}
268
269// ProviderSettings returns the settings used by subsequently spawned
270// children. It is primarily useful to host integrations that need to verify
271// the manager follows the active provider selection.
272func (f *Supervisor) ProviderSettings() (string, bool) {
273 f.mu.Lock()
274 defer f.mu.Unlock()
275 return f.cfg.BaseURL, f.cfg.InsecureTLS
276}
277
278// SetActiveSession scopes the dashboard view (and Spawn stamping)
279// to a particular host zut session id. Pass empty to clear the
280// scope and revert to "show every agent" (the original behaviour).
281//
282// Existing in-memory agents keep their SessionID; only the filter
283// applied at snapshot time changes. So swapping the active session
284// with /sessions instantly re-narrows the dashboard without
285// touching any agent state.
286func (f *Supervisor) SetActiveSession(id string) {
287 f.mu.Lock()
288 defer f.mu.Unlock()
289 f.activeSession = id
290}
291
292// ActiveSession returns the current scope, mostly for tests and
293// diagnostics. Empty means "no scope; show everything".
294func (f *Supervisor) ActiveSession() string {
295 f.mu.Lock()
296 defer f.mu.Unlock()
297 return f.activeSession
298}
299
300// agentStateDir is the per-agent state directory laid out as:
301//
302// <root>/agents/<id>/
303// events.jsonl durable event log (runner-owned)
304// session.json persistent agent session (child-owned)
305// meta.json static metadata (id, task)
306//
307// The transient unix socket inbox lives in a probed runtime directory,
308// which may be on a different filesystem from this durable state.
309func (f *Supervisor) agentStateDir(id string) string {
310 return filepath.Join(f.cfg.Root, "agents", id)
311}
312
313func (f *Supervisor) uniqueAgentIDLocked(base string) string {
314 id := base
315 for suffix := 2; ; suffix++ {
316 _, inMemory := f.agents[id]
317 _, onDiskErr := os.Stat(f.agentStateDir(id))
318 if !inMemory && onDiskErr != nil {
319 return id
320 }
321 id = fmt.Sprintf("%s-%d", base, suffix)
322 }
323}
324
325// SpawnRequest configures a Spawn. Only Task is required; the rest
326// are optional. Model + Provider, when set, get baked into the
327// child argv as --model / --provider so the agent runs against the
328// chosen model regardless of the parent's current selection. A named
329// Subagent is passed to the child as --subagent, where it resolves the
330// markdown profile and applies its instructions. Reasoning is passed
331// as --reasoning when set. FastMode overrides the host setting when a
332// selected profile declares a fastMode preference.
333type SpawnRequest struct {
334 Task string
335 Model string // optional override; child resolves default if empty
336 Provider string // optional override; usually paired with Model
337 BaseURL string // optional provider endpoint override
338 // InsecureTLS enables the child to skip TLS verification for its
339 // effective custom provider endpoint.
340 InsecureTLS bool
341 Reasoning string // optional reasoning/thinking level override
342 // FastMode is an optional profile override. A non-nil value enables or
343 // disables fast mode for this child; nil inherits Config.FastMode.
344 FastMode *bool
345 Subagent string // optional named markdown profile
346
347 // WebSearchPolicy is an internal child capability override. Generic
348 // requests normally inherit the supervisor's parent decision; named
349 // requests use Allow only when their profile explicitly lists web_search.
350 WebSearchPolicy WebSearchPolicy
351
352 // ParentID is metadata used for per-parent scheduling limits. A
353 // non-empty RequesterAgentID identifies an untrusted child request;
354 // recursive spawning is rejected in v1.
355 ParentID string
356 BatchID string
357 RootSessionID string
358 RequesterAgentID string
359
360 // Required keeps this delegated turn on the parent's critical path. The
361 // supervisor persists its outcome and the manager tool waits for it.
362 Required bool
363 Timeout time.Duration
364 MaxTurns int
365 Tools []string
366 WorkspaceMode WorkspaceMode
367 WorkspaceBase string
368 WorkspaceCapture CaptureMode
369}
370
371// Spawn creates a new Agent for the given task, allocates its on-disk
372// state directory (events log, inbox socket path, session file path), and
373// queues it for the manager-owned scheduler. The returned Agent is pending
374// when the concurrency budget is full and transitions to running when a slot
375// opens. This is the historical signature; callers that want to override the
376// child's model or workspace use SpawnReq instead.
377func (f *Supervisor) Spawn(ctx context.Context, task string) (*Agent, error) {
378 return f.SpawnReq(ctx, SpawnRequest{Task: task})
379}
380
381// unqualifiedModelID removes a redundant provider prefix from a model ID.
382// Model IDs themselves may contain slashes, so only the explicitly selected
383// provider's exact prefix is removed.
384func unqualifiedModelID(provider, model string) string {
385 provider = strings.TrimSpace(provider)
386 model = strings.TrimSpace(model)
387 if provider == "" {
388 return model
389 }
390 return strings.TrimPrefix(model, provider+"/")
391}
392
393// SpawnReq is the full-fat variant of Spawn that accepts a SpawnRequest.
394// Existing callers can keep using Spawn; new code that wants to pin the
395// child's model, policy, or workspace uses this.
396func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, error) {
397 task := strings.TrimSpace(req.Task)
398 if task == "" {
399 return nil, errors.New("subagents: empty task")
400 }
401 if req.RequesterAgentID != "" {
402 return nil, errors.New("subagents: child-originated spawning is disabled")
403 }
404 if err := f.validateSpawnRequest(req); err != nil {
405 return nil, err
406 }
407 req.Model = unqualifiedModelID(req.Provider, req.Model)
408 if ctx == nil {
409 ctx = context.Background()
410 }
411
412 now := f.cfg.Now()
413 baseID := newAgentID(task, now)
414
415 f.mu.Lock()
416 id := f.uniqueAgentIDLocked(baseID)
417 dir := f.cfg.RepoRoot
418 fastMode := f.cfg.FastMode
419 configProvider := f.cfg.Provider
420 configBaseURL := f.cfg.BaseURL
421 configInsecureTLS := f.cfg.InsecureTLS
422 sessionID := f.activeSession
423 f.mu.Unlock()
424 fastModeOverridesHost := false
425 if req.FastMode != nil {
426 fastModeOverridesHost = !fastMode && *req.FastMode
427 fastMode = *req.FastMode
428 }
429 childProvider := strings.TrimSpace(req.Provider)
430 inheritProviderSettings := childProvider == "" || configProvider == "" || strings.EqualFold(childProvider, strings.TrimSpace(configProvider))
431 baseURL := strings.TrimSpace(req.BaseURL)
432 if baseURL == "" && inheritProviderSettings {
433 baseURL = strings.TrimSpace(configBaseURL)
434 }
435 insecureTLS := req.InsecureTLS || (inheritProviderSettings && configInsecureTLS)
436
437 stateDir := f.agentStateDir(id)
438 if err := os.MkdirAll(stateDir, 0o700); err != nil {
439 return nil, fmt.Errorf("subagents state dir: %w", err)
440 }
441 lease, err := acquireAgentLease(stateDir)
442 if err != nil {
443 return nil, fmt.Errorf("subagents agent lease: %w", err)
444 }
445 leaseOwned := true
446 defer func() {
447 if leaseOwned {
448 _ = lease.Close()
449 }
450 }()
451 logPath := filepath.Join(stateDir, "events.jsonl")
452 sessionPath := filepath.Join(stateDir, "session.json")
453 // Keep the transient unix socket outside the durable state root.
454 // ZUT_HOME may live on a shared or network filesystem that supports
455 // regular state files but cannot host unix socket nodes.
456 inboxPath, err := inboxSocketPath(f.cfg.Root, id)
457 if err != nil {
458 _ = lease.Close()
459 leaseOwned = false
460 _ = os.RemoveAll(stateDir)
461 return nil, fmt.Errorf("subagents inbox path: %w", err)
462 }
463
464 workspaceMode := req.WorkspaceMode
465 if workspaceMode == "" {
466 workspaceMode = WorkspaceShared
467 }
468 workspace, err := PrepareWorkspace(ctx, WorkspaceRequest{
469 Mode: workspaceMode,
470 RepositoryRoot: dir,
471 StateDir: stateDir,
472 AgentID: id,
473 Base: req.WorkspaceBase,
474 Capture: req.WorkspaceCapture,
475 AllowedRoots: append([]string(nil), f.cfg.Policy.AllowedRoots...),
476 })
477 if err != nil {
478 _ = lease.Close()
479 leaseOwned = false
480 _ = os.RemoveAll(stateDir)
481 return nil, err
482 }
483 dir = workspace.Dir()
484 maxTurns := req.MaxTurns
485 if maxTurns <= 0 {
486 maxTurns = f.cfg.Policy.MaxTurns
487 }
488 timeout := req.Timeout
489 if timeout <= 0 {
490 timeout = f.cfg.Policy.DefaultTimeout
491 }
492 runCtx, cancel := f.workerContext()
493 tools := append([]string(nil), req.Tools...)
494 if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 {
495 tools = append([]string(nil), f.cfg.Policy.AllowedTools...)
496 }
497 webSearchPolicy := f.resolveWebSearchPolicy(req)
498
499 a := &Agent{
500 ID: id,
501 Task: task,
502 OriginalTask: task,
503 Dir: dir,
504 RepositoryRoot: workspace.RepositoryRoot(),
505 WorkspacePath: workspace.Dir(),
506 Started: now,
507 ParentID: strings.TrimSpace(req.ParentID),
508 BatchID: strings.TrimSpace(req.BatchID),
509 RootSessionID: strings.TrimSpace(req.RootSessionID),
510 Model: strings.TrimSpace(req.Model),
511 Provider: strings.TrimSpace(req.Provider),
512 BaseURL: baseURL,
513 InsecureTLS: insecureTLS,
514 Reasoning: strings.TrimSpace(req.Reasoning),
515 FastMode: fastMode,
516 Subagent: strings.TrimSpace(req.Subagent),
517 SessionID: sessionID,
518 WorkspaceMode: workspaceMode,
519 WorkspaceBase: strings.TrimSpace(req.WorkspaceBase),
520 WorkspaceCapture: req.WorkspaceCapture,
521 MaxTurns: maxTurns,
522 Timeout: timeout,
523 HeartbeatInterval: f.cfg.Policy.HeartbeatInterval,
524 Tools: tools,
525 WebSearchPolicy: webSearchPolicy,
526 InboxPath: inboxPath,
527 EventLogPath: logPath,
528 SessionPath: sessionPath,
529 inbox: NewInbox(inboxPath),
530 status: StatusPending,
531 activity: "queued",
532 transcriptLoaded: true,
533 processState: ProcessPending,
534 turnState: TurnQueued,
535 updatedAt: now,
536 lastActivity: now,
537 requirement: newRequirement(req.Required, 1, now),
538 stateDir: stateDir,
539 lease: lease,
540 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
541 maxOutputLines: f.cfg.Policy.MaxOutputLines,
542 done: make(chan struct{}),
543 turnResults: make(chan *TurnResult, 16),
544 }
545 a.fastModeOverridesHost = fastModeOverridesHost
546 if a.RootSessionID == "" {
547 a.RootSessionID = sessionID
548 }
549 a.ctx, a.cancel = runCtx, cancel
550 a.persistFn = f.persistAgent
551 a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
552 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
553 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
554 a.runner = f.cfg.NewRunner(a)
555 if err := writeAgentMeta(stateDir, a); err != nil {
556 if a.cancel != nil {
557 a.cancel()
558 }
559 _ = workspace.Cleanup(context.Background())
560 _ = lease.Close()
561 leaseOwned = false
562 _ = os.RemoveAll(stateDir)
563 return nil, fmt.Errorf("subagents initial metadata: %w", err)
564 }
565
566 f.mu.Lock()
567 f.agents[id] = a
568 f.order = append(f.order, id)
569 f.queue = append(f.queue, a)
570 f.mu.Unlock()
571 leaseOwned = false
572
573 // Persisted metadata was written before queue admission, so a later
574 // supervisor can reconstruct work that never reached a process.
575 f.armQueueTimeout(a)
576 f.schedule()
577 return a, nil
578}
579
580// SendUserTurn is sugar for the common "send the next user turn"
581// case. It quotes nothing and forwards verbatim; callers are
582// expected to have already trimmed and expanded the text.
583// CancelTurn requests cancellation of the current turn while keeping the
584// worker process alive for a later follow-up.
585func (f *Supervisor) CancelTurn(id string) error {
586 a := f.Get(id)
587 if a == nil {
588 return fmt.Errorf("subagents: no such agent %q", id)
589 }
590 if a.inbox == nil {
591 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
592 }
593 a.setTurnState(TurnCanceling, a.CurrentTurnID())
594 return a.inbox.SendCommand(NewCommand(CommandTurnCancel, a.ID, a.CurrentTurnID(), TurnCancelPayload{Reason: "user"}))
595}
596
597func (f *Supervisor) SendUserTurn(id, text string) error {
598 a := f.Get(id)
599 if a == nil {
600 return fmt.Errorf("subagents: no such agent %q", id)
601 }
602 if a.inbox == nil {
603 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
604 }
605 return a.inbox.SendCommand(NewCommand(CommandTurnStart, a.ID, a.CurrentTurnID(), TurnStartPayload{Prompt: text}))
606}
607
608func (f *Supervisor) run(a *Agent) {
609 a.mu.Lock()
610 killedBeforeStart := a.status == StatusKilled
611 ctxErr := error(nil)
612 if a.ctx != nil {
613 ctxErr = a.ctx.Err()
614 }
615 a.mu.Unlock()
616 var err error
617 if killedBeforeStart || ctxErr != nil || !atomic.CompareAndSwapInt32(&a.launchState, 0, 1) {
618 if ctxErr != nil {
619 err = ctxErr
620 } else {
621 err = context.Canceled
622 }
623 } else {
624 a.setActivity("starting")
625 if a.runner == nil {
626 err = errors.New("subagents: agent has no runner")
627 } else {
628 err = a.runner.Run(a.ctx, agentSink{a: a})
629 }
630 }
631
632 a.mu.Lock()
633 a.finished = f.cfg.Now()
634 switch {
635 case a.status == StatusKilled:
636 // Stop requested a terminal state; preserve it.
637 case errors.Is(err, context.Canceled):
638 a.status = StatusKilled
639 a.activity = "cancelled"
640 a.lastErr = err
641 case errors.Is(err, context.DeadlineExceeded):
642 a.status = StatusFailed
643 a.activity = "timeout"
644 a.lastErr = err
645 case err != nil:
646 a.status = StatusFailed
647 a.activity = "error: " + truncate(err.Error(), 120)
648 a.lastErr = err
649 default:
650 a.status = StatusDone
651 a.activity = "done"
652 }
653 finalStatus := a.status
654 a.mu.Unlock()
655
656 switch finalStatus {
657 case StatusKilled:
658 a.setProcessState(ProcessKilled)
659 a.setTurnState(TurnCanceled, a.CurrentTurnID())
660 case StatusFailed:
661 a.setProcessState(ProcessExited)
662 a.setTurnState(TurnFailed, a.CurrentTurnID())
663 default:
664 a.setProcessState(ProcessExited)
665 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
666 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
667 }
668 }
669
670 // The process slot is no longer occupied once Runner.Run has returned.
671 // Release it before potentially slow artifact capture and workspace
672 // cleanup so queued work can start without waiting on git/filesystem I/O.
673 f.releaseCapacity(a)
674
675 // Capture isolated output and write the structured result before the
676 // temporary workspace is removed. This ordering is the recovery
677 // invariant: a user can always inspect a durable result after failure.
678 f.captureWorkspace(a)
679 f.ensureResult(a, finalStatus, err)
680 if a.cancel != nil {
681 a.cancel()
682 }
683 if a.inbox != nil {
684 _ = a.inbox.Close()
685 }
686 a.setProcessPID(0)
687 f.persistAgent(a)
688 if a.workspaceCleanup != nil {
689 _ = a.workspaceCleanup()
690 }
691 _ = a.releaseLease()
692 a.closeDone()
693}
694
695// List returns a snapshot of every agent in creation order. The
696// returned slice is a copy; callers may iterate without holding the
697// subagent lock. Agent fields are read under their own mutex during
698// formatting in Snapshot.
699func (f *Supervisor) List() []*Agent {
700 f.mu.Lock()
701 defer f.mu.Unlock()
702 out := make([]*Agent, 0, len(f.order))
703 for _, id := range f.order {
704 out = append(out, f.agents[id])
705 }
706 return out
707}
708
709// Get returns the agent with the given (possibly truncated) id, or
710// nil. Matching is prefix-based so the user can type the first few
711// characters of a long id.
712func (f *Supervisor) Get(id string) *Agent {
713 f.mu.Lock()
714 defer f.mu.Unlock()
715 if a, ok := f.agents[id]; ok {
716 return a
717 }
718 // Prefix match.
719 var hits []*Agent
720 for k, a := range f.agents {
721 if strings.HasPrefix(k, id) {
722 hits = append(hits, a)
723 }
724 }
725 if len(hits) == 1 {
726 return hits[0]
727 }
728 return nil
729}
730
731var errDetachedStopTimeout = errors.New("detached worker stop timeout")
732
733const detachedStopPollInterval = 20 * time.Millisecond
734
735// waitForDetachedWorker waits for the worker inbox to disappear, returning a
736// sentinel when the grace period expires. Polling is interruptible so a
737// caller's context is not made to wait for the next fixed sleep interval.
738func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
739 if ctx == nil {
740 ctx = context.Background()
741 }
742 timer := time.NewTimer(timeout)
743 defer timer.Stop()
744 ticker := time.NewTicker(detachedStopPollInterval)
745 defer ticker.Stop()
746 for {
747 if err := ctx.Err(); err != nil {
748 return err
749 }
750 if !inboxLive(path) {
751 return nil
752 }
753 select {
754 case <-ctx.Done():
755 return ctx.Err()
756 case <-timer.C:
757 if !inboxLive(path) {
758 return nil
759 }
760 return errDetachedStopTimeout
761 case <-ticker.C:
762 }
763 }
764}
765
766// Stop cancels the agent's context. The Runner should observe the
767// cancellation and return promptly; the goroutine then finalises the
768// agent in StatusKilled. Also closes the supervisor-side inbox handle
769// so any pending send retries fail fast instead of dialing a
770// socket about to be unlinked. Stop uses a background context; callers
771// that need cancellation while waiting for a detached worker should use
772// StopContext.
773//
774// Stop is a no-op for any agent that's not in a live runnable state
775// — Done / Failed / Killed (already finalised). Reloaded Detached agents
776// are probed first: if their worker survived the supervisor, Stop sends the
777// shutdown command through the durable inbox instead of pretending no
778// process exists.
779func (f *Supervisor) Stop(id string) error {
780 return f.stop(context.Background(), id, ShutdownOriginTargeted)
781}
782
783// StopContext is Stop with a context that can interrupt detached-worker
784// shutdown and its graceful and force-stop waits. The context is not held
785// under the supervisor operation lock while those waits run.
786func (f *Supervisor) StopContext(ctx context.Context, id string) error {
787 return f.stop(ctx, id, ShutdownOriginTargeted)
788}
789
790func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
791 if ctx == nil {
792 ctx = context.Background()
793 }
794 f.operationMu.Lock()
795 a := f.Get(id)
796 if a == nil {
797 f.operationMu.Unlock()
798 return fmt.Errorf("no such agent %q", id)
799 }
800
801 // Coordinate the status transition with schedule. Holding f.mu while
802 // checking and changing a.status closes the small window where Stop could
803 // observe pending, schedule could promote the same agent to running, and
804 // the pending cleanup below could then finalize an agent that was already
805 // admitted to the runner.
806 f.mu.Lock()
807 status, shutdownStarted := a.prepareShutdown(origin)
808 if !shutdownStarted {
809 f.mu.Unlock()
810 f.operationMu.Unlock()
811 return nil
812 }
813 if status == StatusDetached {
814 path := a.InboxPath
815 inbox := a.inbox
816 pid := a.ProcessPIDValue()
817 f.mu.Unlock()
818 // The detached worker may take the full grace period to exit. Do not
819 // serialize unrelated supervisor operations for that entire wait.
820 f.operationMu.Unlock()
821 return f.stopDetached(ctx, a, path, inbox, pid, origin)
822 }
823 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
824 if status == StatusPending {
825 for i, queued := range f.queue {
826 if queued == a {
827 f.queue = append(f.queue[:i], f.queue[i+1:]...)
828 break
829 }
830 }
831 }
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.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
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, origin ShutdownOrigin) 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.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); 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.ensureResult(a, StatusKilled, context.Canceled)
970 f.persistAgent(a)
971 a.closeDone()
972}
973
974// StopByRootSession stops and waits for every agent owned by rootSessionID.
975// It leaves the supervisor lifetime and agents from other sessions running.
976func (f *Supervisor) StopByRootSession(rootSessionID string) error {
977 return f.StopByRootSessionContext(context.Background(), rootSessionID)
978}
979
980// StopByRootSessionContext is the context-aware form of StopByRootSession.
981func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
982 if ctx == nil {
983 ctx = context.Background()
984 }
985 rootSessionID = strings.TrimSpace(rootSessionID)
986 if rootSessionID == "" {
987 return errors.New("subagents: root session id is required")
988 }
989 agents := f.List()
990 owned := make([]*Agent, 0, len(agents))
991 for _, a := range agents {
992 if a.RootSessionID == rootSessionID {
993 owned = append(owned, a)
994 }
995 }
996 return f.stopAgents(ctx, owned, ShutdownOriginSession)
997}
998
999func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
1000 var firstErr error
1001 for _, a := range agents {
1002 if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
1003 firstErr = err
1004 }
1005 }
1006 for _, a := range agents {
1007 select {
1008 case <-a.done:
1009 case <-ctx.Done():
1010 if firstErr == nil {
1011 firstErr = ctx.Err()
1012 }
1013 return firstErr
1014 }
1015 }
1016 return firstErr
1017}
1018
1019// StopAll cancels every running agent. Used on shutdown.
1020func (f *Supervisor) StopAll() {
1021 _ = f.stopAll(context.Background())
1022}
1023
1024// StopAllContext is the context-aware shutdown variant. It returns the
1025// context error when cancellation interrupts a detached-worker wait.
1026func (f *Supervisor) StopAllContext(ctx context.Context) error {
1027 return f.stopAll(ctx)
1028}
1029
1030func (f *Supervisor) stopAll(ctx context.Context) error {
1031 if ctx == nil {
1032 ctx = context.Background()
1033 }
1034 defer func() {
1035 if f.lifetimeCancel != nil {
1036 f.lifetimeCancel()
1037 }
1038 }()
1039 return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
1040}
1041
1042// Remove tears down the per-agent state for a terminated agent. It
1043// is an error to remove an agent that's still running; call Stop
1044// first and wait for the status to settle. Detached agents
1045// (reloaded from disk) remove cleanly because they have no live
1046// runner racing for the same files.
1047//
1048// Shared-workspace agents never touch source files during Remove; worktree
1049// agents have already captured their output and removed the isolated checkout
1050// during finalization. Remove then deletes the agent's durable state directory
1051// under <root>/agents/<id>/.
1052func (f *Supervisor) Remove(id string) error {
1053 f.operationMu.Lock()
1054 defer f.operationMu.Unlock()
1055 a := f.Get(id)
1056 if a == nil {
1057 return fmt.Errorf("no such agent %q", id)
1058 }
1059 a.mu.Lock()
1060 st := a.status
1061 inboxPath := a.InboxPath
1062 a.mu.Unlock()
1063 if st == StatusRunning || st == StatusPending {
1064 return fmt.Errorf("agent %s still %s", a.ID, st)
1065 }
1066 if inboxLive(inboxPath) {
1067 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1068 }
1069 select {
1070 case <-a.done:
1071 default:
1072 return fmt.Errorf("agent %s is still finalizing", a.ID)
1073 }
1074 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1075 if err != nil {
1076 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1077 }
1078 _ = lease.Close()
1079 // Best-effort cleanup of the per-agent state directory
1080 // (meta.json, events.jsonl, session.json). Failing here would leave
1081 // the user with no recourse,
1082 // so swallow the error.
1083 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1084 f.mu.Lock()
1085 delete(f.agents, a.ID)
1086 for i, k := range f.order {
1087 if k == a.ID {
1088 f.order = append(f.order[:i], f.order[i+1:]...)
1089 break
1090 }
1091 }
1092 f.mu.Unlock()
1093 return nil
1094}
1095
1096// Snapshot returns a read-only view of one agent. Safe for the TUI
1097// goroutine to call repeatedly; never blocks on the Runner.
1098type AgentSnapshot struct {
1099 ID string
1100 Task string
1101 Dir string
1102 Status Status
1103 ProcessState ProcessState
1104 TurnState TurnState
1105 CurrentTurnID string
1106 LifetimeTurns int
1107 CurrentRunTurns int
1108 Attempt int
1109 Activity string
1110 Started time.Time
1111 Finished time.Time
1112 CreatedAt time.Time
1113 UpdatedAt time.Time
1114 LastActivity time.Time
1115 Err string
1116 Tail string // last few transcript lines, joined with "\n"
1117 Lines []string // full transcript (already capped by Agent.appendTranscript)
1118 LastAssistant string // complete final assistant message, without role formatting
1119 Result *TurnResult
1120 ResultRef string
1121 Requirement RequirementSnapshot
1122 PatchRef string
1123 ChangedFiles []string
1124 OutputTruncated bool
1125
1126 // Model, Provider, Reasoning, and the provider connection settings
1127 // expose the per-agent configuration captured at Spawn time. Empty
1128 // values inherit the child's defaults. Subagent is the selected named
1129 // markdown profile. The dashboard surfaces these so the user can
1130 // confirm the child configuration.
1131 Model string
1132 Provider string
1133 BaseURL string
1134 InsecureTLS bool
1135 Reasoning string
1136 FastMode bool
1137 Subagent string
1138 WorkspaceMode WorkspaceMode
1139 WorkspacePath string
1140
1141 // Paths to the agent's durable state. Surface them in the
1142 // snapshot so the dashboard /subagents view can
1143 // read events.jsonl or resume the session without going back through the Agent.
1144 InboxPath string
1145 EventLogPath string
1146 SessionPath string
1147}
1148
1149// Snapshot copies the live agent state into a value the caller can
1150// inspect at leisure.
1151func (a *Agent) Snapshot() AgentSnapshot {
1152 a.mu.Lock()
1153 tail := strings.Join(lastN(a.transcript, 6), "\n")
1154 lines := make([]string, len(a.transcript))
1155 copy(lines, a.transcript)
1156 errStr := ""
1157 if a.lastErr != nil {
1158 errStr = a.lastErr.Error()
1159 }
1160 status := a.status
1161 activity := a.activity
1162 lastAssistant := a.lastAssistant
1163 outputTruncated := a.outputTruncated || a.streamingAssistantTruncated
1164 finished := a.finished
1165 a.mu.Unlock()
1166
1167 a.lifecycleMu.Lock()
1168 processState := a.processState
1169 turnState := a.turnState
1170 currentTurnID := a.currentTurnID
1171 lifetimeTurns := a.LifetimeTurns
1172 currentRunTurns := a.CurrentRunTurns
1173 attempt := a.Attempt
1174 lastActivity := a.lastActivity
1175 updatedAt := a.updatedAt
1176 result := cloneTurnResult(a.result)
1177 resultRef := a.resultRef
1178 requirement := a.visibleRequirementLocked()
1179 patchRef := a.patchRef
1180 changedFiles := append([]string(nil), a.changedFiles...)
1181 a.lifecycleMu.Unlock()
1182 return AgentSnapshot{
1183 ID: a.ID, Task: a.Task, Dir: a.Dir,
1184 Status: status, ProcessState: processState, TurnState: turnState,
1185 CurrentTurnID: currentTurnID, LifetimeTurns: lifetimeTurns, CurrentRunTurns: currentRunTurns, Attempt: attempt, Activity: activity,
1186 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1187 Err: errStr, Tail: tail, Lines: lines,
1188 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1189 Requirement: requirement, PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1190 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1191 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1192 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1193 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1194 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1195 }
1196}
1197
1198// SnapshotAll returns snapshots of every agent in creation order,
1199// scoped to the active session when one is set.
1200//
1201// Scoping rules:
1202// - activeSession == "": no filter, every agent is returned
1203// (historical behaviour; used by tests and scripted callers).
1204// - activeSession != "": include live agents whose SessionID matches
1205// activeSession, plus unscoped agents. Detached and terminal
1206// agents from another host session remain visible so restart/recovery
1207// does not hide durable work.
1208func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1209 agents := f.List()
1210 f.mu.Lock()
1211 active := f.activeSession
1212 f.mu.Unlock()
1213
1214 out := make([]AgentSnapshot, 0, len(agents))
1215 for _, a := range agents {
1216 if active != "" && a.SessionID != "" && a.SessionID != active {
1217 // Keep detached/terminal entries visible after a normal host
1218 // restart so durable recovery remains discoverable. Hide only
1219 // live work belonging to another in-process host session.
1220 status := a.Status()
1221 if status == StatusRunning || status == StatusPending {
1222 continue
1223 }
1224 }
1225 out = append(out, a.Snapshot())
1226 }
1227 // Sort by start time for a stable, deterministic listing.
1228 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1229 return out
1230}
1231
1232// agentSink is the Sink the Supervisor hands to each Runner.
1233type agentSink struct{ a *Agent }
1234
1235func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1236func (s agentSink) Transcript(chunk string) {
1237 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") {
1238 s.a.appendTranscript(chunk)
1239 return
1240 }
1241 s.a.appendAssistantMessage(chunk)
1242}
1243func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1244func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1245func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
1246func (s agentSink) resetStreamingAssistant() { s.a.resetStreamingAssistant() }
1247
1248func truncate(s string, n int) string {
1249 runes := []rune(s)
1250 if len(runes) <= n {
1251 return s
1252 }
1253 if n <= 3 {
1254 return strings.Repeat(".", n)
1255 }
1256 return string(runes[:n-3]) + "..."
1257}
1258
1259func lastN(lines []string, n int) []string {
1260 if len(lines) <= n {
1261 return lines
1262 }
1263 return lines[len(lines)-n:]
1264}