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 completion path. The
361 // supervisor persists its outcome while the manager remains asynchronous.
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 startupErr := a.startupErr
635 terminalErr := err
636 switch {
637 case a.status == StatusKilled:
638 // Stop requested a terminal state; preserve it.
639 case startupErr != nil:
640 a.status = StatusFailed
641 a.activity = "startup timeout: worker did not become ready"
642 a.lastErr = startupErr
643 terminalErr = startupErr
644 case errors.Is(err, context.Canceled):
645 a.status = StatusKilled
646 a.activity = "cancelled"
647 a.lastErr = err
648 case errors.Is(err, context.DeadlineExceeded):
649 a.status = StatusFailed
650 a.activity = "timeout"
651 a.lastErr = err
652 case err != nil:
653 a.status = StatusFailed
654 a.activity = "error: " + truncate(err.Error(), 120)
655 a.lastErr = err
656 default:
657 a.status = StatusDone
658 a.activity = "done"
659 }
660 finalStatus := a.status
661 a.mu.Unlock()
662
663 switch finalStatus {
664 case StatusKilled:
665 a.setProcessState(ProcessKilled)
666 a.setTurnState(TurnCanceled, a.CurrentTurnID())
667 case StatusFailed:
668 a.setProcessState(ProcessExited)
669 a.setTurnState(TurnFailed, a.CurrentTurnID())
670 default:
671 a.setProcessState(ProcessExited)
672 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
673 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
674 }
675 }
676
677 // The process slot is no longer occupied once Runner.Run has returned.
678 // Release it before potentially slow artifact capture and workspace
679 // cleanup so queued work can start without waiting on git/filesystem I/O.
680 f.releaseCapacity(a)
681
682 // Capture isolated output and write the structured result before the
683 // temporary workspace is removed. This ordering is the recovery
684 // invariant: a user can always inspect a durable result after failure.
685 f.captureWorkspace(a)
686 f.ensureResult(a, finalStatus, terminalErr)
687 if a.cancel != nil {
688 a.cancel()
689 }
690 if a.inbox != nil {
691 _ = a.inbox.Close()
692 }
693 a.setProcessPID(0)
694 f.persistAgent(a)
695 if a.workspaceCleanup != nil {
696 _ = a.workspaceCleanup()
697 }
698 _ = a.releaseLease()
699 a.closeDone()
700}
701
702// List returns a snapshot of every agent in creation order. The
703// returned slice is a copy; callers may iterate without holding the
704// subagent lock. Agent fields are read under their own mutex during
705// formatting in Snapshot.
706func (f *Supervisor) List() []*Agent {
707 f.mu.Lock()
708 defer f.mu.Unlock()
709 out := make([]*Agent, 0, len(f.order))
710 for _, id := range f.order {
711 out = append(out, f.agents[id])
712 }
713 return out
714}
715
716// Get returns the agent with the given (possibly truncated) id, or
717// nil. Matching is prefix-based so the user can type the first few
718// characters of a long id.
719func (f *Supervisor) Get(id string) *Agent {
720 f.mu.Lock()
721 defer f.mu.Unlock()
722 if a, ok := f.agents[id]; ok {
723 return a
724 }
725 // Prefix match.
726 var hits []*Agent
727 for k, a := range f.agents {
728 if strings.HasPrefix(k, id) {
729 hits = append(hits, a)
730 }
731 }
732 if len(hits) == 1 {
733 return hits[0]
734 }
735 return nil
736}
737
738var errDetachedStopTimeout = errors.New("detached worker stop timeout")
739
740const detachedStopPollInterval = 20 * time.Millisecond
741
742// waitForDetachedWorker waits for the worker inbox to disappear, returning a
743// sentinel when the grace period expires. Polling is interruptible so a
744// caller's context is not made to wait for the next fixed sleep interval.
745func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
746 if ctx == nil {
747 ctx = context.Background()
748 }
749 timer := time.NewTimer(timeout)
750 defer timer.Stop()
751 ticker := time.NewTicker(detachedStopPollInterval)
752 defer ticker.Stop()
753 for {
754 if err := ctx.Err(); err != nil {
755 return err
756 }
757 if !inboxLive(path) {
758 return nil
759 }
760 select {
761 case <-ctx.Done():
762 return ctx.Err()
763 case <-timer.C:
764 if !inboxLive(path) {
765 return nil
766 }
767 return errDetachedStopTimeout
768 case <-ticker.C:
769 }
770 }
771}
772
773// Stop cancels the agent's context. The Runner should observe the
774// cancellation and return promptly; the goroutine then finalises the
775// agent in StatusKilled. Also closes the supervisor-side inbox handle
776// so any pending send retries fail fast instead of dialing a
777// socket about to be unlinked. Stop uses a background context; callers
778// that need cancellation while waiting for a detached worker should use
779// StopContext.
780//
781// Stop is a no-op for any agent that's not in a live runnable state
782// — Done / Failed / Killed (already finalised). Reloaded Detached agents
783// are probed first: if their worker survived the supervisor, Stop sends the
784// shutdown command through the durable inbox instead of pretending no
785// process exists.
786func (f *Supervisor) Stop(id string) error {
787 return f.stop(context.Background(), id, ShutdownOriginTargeted)
788}
789
790// StopContext is Stop with a context that can interrupt detached-worker
791// shutdown and its graceful and force-stop waits. The context is not held
792// under the supervisor operation lock while those waits run.
793func (f *Supervisor) StopContext(ctx context.Context, id string) error {
794 return f.stop(ctx, id, ShutdownOriginTargeted)
795}
796
797func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
798 if ctx == nil {
799 ctx = context.Background()
800 }
801 f.operationMu.Lock()
802 a := f.Get(id)
803 if a == nil {
804 f.operationMu.Unlock()
805 return fmt.Errorf("no such agent %q", id)
806 }
807
808 // Coordinate the status transition with schedule. Holding f.mu while
809 // checking and changing a.status closes the small window where Stop could
810 // observe pending, schedule could promote the same agent to running, and
811 // the pending cleanup below could then finalize an agent that was already
812 // admitted to the runner.
813 f.mu.Lock()
814 status, shutdownStarted := a.prepareShutdown(origin)
815 if !shutdownStarted {
816 f.mu.Unlock()
817 f.operationMu.Unlock()
818 return nil
819 }
820 if status == StatusDetached {
821 path := a.InboxPath
822 inbox := a.inbox
823 pid := a.ProcessPIDValue()
824 f.mu.Unlock()
825 // The detached worker may take the full grace period to exit. Do not
826 // serialize unrelated supervisor operations for that entire wait.
827 f.operationMu.Unlock()
828 return f.stopDetached(ctx, a, path, inbox, pid, origin)
829 }
830 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
831 if status == StatusPending {
832 for i, queued := range f.queue {
833 if queued == a {
834 f.queue = append(f.queue[:i], f.queue[i+1:]...)
835 break
836 }
837 }
838 }
839 f.mu.Unlock()
840 f.operationMu.Unlock()
841 a.setTurnState(TurnCanceling, a.CurrentTurnID())
842
843 if status == StatusPending {
844 a.setProcessState(ProcessKilled)
845 a.setTurnState(TurnCanceled, a.CurrentTurnID())
846 f.captureWorkspace(a)
847 f.ensureResult(a, StatusKilled, context.Canceled)
848 if a.cancel != nil {
849 a.cancel()
850 }
851 f.persistAgent(a)
852 if a.workspaceCleanup != nil {
853 _ = a.workspaceCleanup()
854 }
855 _ = a.releaseLease()
856 a.closeDone()
857 f.schedule()
858 if err := ctx.Err(); err != nil {
859 return err
860 }
861 return nil
862 }
863
864 // Ask a live worker to shut down cleanly first. If the socket is not
865 // ready, fall back immediately; a forceful context cancellation is
866 // safer than leaving a supervisor entry running forever.
867 if err := ctx.Err(); err != nil {
868 if a.cancel != nil {
869 a.cancel()
870 }
871 return err
872 }
873 graceful := false
874 if a.inbox != nil {
875 err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
876 if err == nil {
877 graceful = true
878 }
879 }
880 if !graceful && a.cancel != nil {
881 a.cancel()
882 }
883 if graceful {
884 grace := f.cfg.Policy.CancelGracePeriod
885 if grace <= 0 {
886 grace = 10 * time.Second
887 }
888 go func() {
889 timer := time.NewTimer(grace)
890 defer timer.Stop()
891 select {
892 case <-a.done:
893 case <-timer.C:
894 if a.cancel != nil {
895 a.cancel()
896 }
897 if a.inbox != nil {
898 _ = a.inbox.Close()
899 }
900 }
901 }()
902 }
903 if err := ctx.Err(); err != nil {
904 if a.cancel != nil {
905 a.cancel()
906 }
907 if a.inbox != nil {
908 _ = a.inbox.Close()
909 }
910 return err
911 }
912 return nil
913}
914
915func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int, origin ShutdownOrigin) error {
916 if inbox == nil {
917 return nil
918 }
919 if err := ctx.Err(); err != nil {
920 return err
921 }
922 if !inboxLive(path) {
923 return nil
924 }
925 if err := ctx.Err(); err != nil {
926 return err
927 }
928 if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); err != nil {
929 if ctxErr := ctx.Err(); ctxErr != nil {
930 return ctxErr
931 }
932 return err
933 }
934 a.setActivity("shutdown requested")
935 grace := f.cfg.Policy.CancelGracePeriod
936 if grace <= 0 {
937 grace = 10 * time.Second
938 }
939 waitErr := waitForDetachedWorker(ctx, path, grace)
940 if waitErr != nil && !errors.Is(waitErr, errDetachedStopTimeout) {
941 return waitErr
942 }
943 if waitErr == nil {
944 f.markDetachedKilled(a)
945 return nil
946 }
947 if err := ctx.Err(); err != nil {
948 return err
949 }
950 if pid <= 0 {
951 return fmt.Errorf("agent %s live worker did not stop within %s and has no recorded pid", a.ID, grace)
952 }
953 if killErr := forceKillProcess(pid); killErr != nil {
954 return fmt.Errorf("agent %s force-stop: %w", a.ID, killErr)
955 }
956 if waitErr := waitForDetachedWorker(ctx, path, time.Second); waitErr != nil {
957 if errors.Is(waitErr, errDetachedStopTimeout) {
958 return fmt.Errorf("agent %s live worker did not stop after force-stop", a.ID)
959 }
960 return waitErr
961 }
962 f.markDetachedKilled(a)
963 return nil
964}
965
966func (f *Supervisor) markDetachedKilled(a *Agent) {
967 f.operationMu.Lock()
968 defer f.operationMu.Unlock()
969 a.mu.Lock()
970 a.status = StatusKilled
971 a.activity = "killed"
972 a.mu.Unlock()
973 a.setProcessState(ProcessKilled)
974 a.setTurnState(TurnCanceled, a.CurrentTurnID())
975 a.setProcessPID(0)
976 f.ensureResult(a, StatusKilled, context.Canceled)
977 f.persistAgent(a)
978 a.closeDone()
979}
980
981// StopByRootSession stops and waits for every agent owned by rootSessionID.
982// It leaves the supervisor lifetime and agents from other sessions running.
983func (f *Supervisor) StopByRootSession(rootSessionID string) error {
984 return f.StopByRootSessionContext(context.Background(), rootSessionID)
985}
986
987// StopByRootSessionContext is the context-aware form of StopByRootSession.
988func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
989 if ctx == nil {
990 ctx = context.Background()
991 }
992 rootSessionID = strings.TrimSpace(rootSessionID)
993 if rootSessionID == "" {
994 return errors.New("subagents: root session id is required")
995 }
996 agents := f.List()
997 owned := make([]*Agent, 0, len(agents))
998 for _, a := range agents {
999 if a.RootSessionID == rootSessionID {
1000 owned = append(owned, a)
1001 }
1002 }
1003 return f.stopAgents(ctx, owned, ShutdownOriginSession)
1004}
1005
1006func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
1007 var firstErr error
1008 for _, a := range agents {
1009 if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
1010 firstErr = err
1011 }
1012 }
1013 for _, a := range agents {
1014 select {
1015 case <-a.done:
1016 case <-ctx.Done():
1017 if firstErr == nil {
1018 firstErr = ctx.Err()
1019 }
1020 return firstErr
1021 }
1022 }
1023 return firstErr
1024}
1025
1026// StopAll cancels every running agent. Used on shutdown.
1027func (f *Supervisor) StopAll() {
1028 _ = f.stopAll(context.Background())
1029}
1030
1031// StopAllContext is the context-aware shutdown variant. It returns the
1032// context error when cancellation interrupts a detached-worker wait.
1033func (f *Supervisor) StopAllContext(ctx context.Context) error {
1034 return f.stopAll(ctx)
1035}
1036
1037func (f *Supervisor) stopAll(ctx context.Context) error {
1038 if ctx == nil {
1039 ctx = context.Background()
1040 }
1041 defer func() {
1042 if f.lifetimeCancel != nil {
1043 f.lifetimeCancel()
1044 }
1045 }()
1046 return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
1047}
1048
1049// Remove tears down the per-agent state for a terminated agent. It
1050// is an error to remove an agent that's still running; call Stop
1051// first and wait for the status to settle. Detached agents
1052// (reloaded from disk) remove cleanly because they have no live
1053// runner racing for the same files.
1054//
1055// Shared-workspace agents never touch source files during Remove; worktree
1056// agents have already captured their output and removed the isolated checkout
1057// during finalization. Remove then deletes the agent's durable state directory
1058// under <root>/agents/<id>/.
1059func (f *Supervisor) Remove(id string) error {
1060 f.operationMu.Lock()
1061 defer f.operationMu.Unlock()
1062 a := f.Get(id)
1063 if a == nil {
1064 return fmt.Errorf("no such agent %q", id)
1065 }
1066 a.mu.Lock()
1067 st := a.status
1068 inboxPath := a.InboxPath
1069 a.mu.Unlock()
1070 if st == StatusRunning || st == StatusPending {
1071 return fmt.Errorf("agent %s still %s", a.ID, st)
1072 }
1073 if inboxLive(inboxPath) {
1074 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1075 }
1076 select {
1077 case <-a.done:
1078 default:
1079 return fmt.Errorf("agent %s is still finalizing", a.ID)
1080 }
1081 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1082 if err != nil {
1083 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1084 }
1085 _ = lease.Close()
1086 // Best-effort cleanup of the per-agent state directory
1087 // (meta.json, events.jsonl, session.json). Failing here would leave
1088 // the user with no recourse,
1089 // so swallow the error.
1090 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1091 f.mu.Lock()
1092 delete(f.agents, a.ID)
1093 for i, k := range f.order {
1094 if k == a.ID {
1095 f.order = append(f.order[:i], f.order[i+1:]...)
1096 break
1097 }
1098 }
1099 f.mu.Unlock()
1100 return nil
1101}
1102
1103// Snapshot returns a read-only view of one agent. Safe for the TUI
1104// goroutine to call repeatedly; never blocks on the Runner.
1105type AgentSnapshot struct {
1106 ID string
1107 Task string
1108 Dir string
1109 Status Status
1110 ProcessState ProcessState
1111 TurnState TurnState
1112 CurrentTurnID string
1113 LifetimeTurns int
1114 CurrentRunTurns int
1115 Attempt int
1116 Activity string
1117 Started time.Time
1118 Finished time.Time
1119 CreatedAt time.Time
1120 UpdatedAt time.Time
1121 LastActivity time.Time
1122 Err string
1123 Tail string // last few transcript lines, joined with "\n"
1124 Lines []string // full transcript (already capped by Agent.appendTranscript)
1125 LastAssistant string // complete final assistant message, without role formatting
1126 Result *TurnResult
1127 ResultRef string
1128 Requirement RequirementSnapshot
1129 PatchRef string
1130 ChangedFiles []string
1131 OutputTruncated bool
1132
1133 // Model, Provider, Reasoning, and the provider connection settings
1134 // expose the per-agent configuration captured at Spawn time. Empty
1135 // values inherit the child's defaults. Subagent is the selected named
1136 // markdown profile. The dashboard surfaces these so the user can
1137 // confirm the child configuration.
1138 Model string
1139 Provider string
1140 BaseURL string
1141 InsecureTLS bool
1142 Reasoning string
1143 FastMode bool
1144 Subagent string
1145 WorkspaceMode WorkspaceMode
1146 WorkspacePath string
1147
1148 // Paths to the agent's durable state. Surface them in the
1149 // snapshot so the dashboard /subagents view can
1150 // read events.jsonl or resume the session without going back through the Agent.
1151 InboxPath string
1152 EventLogPath string
1153 SessionPath string
1154}
1155
1156// Snapshot copies the live agent state into a value the caller can
1157// inspect at leisure.
1158func (a *Agent) Snapshot() AgentSnapshot {
1159 a.mu.Lock()
1160 tail := strings.Join(lastN(a.transcript, 6), "\n")
1161 lines := make([]string, len(a.transcript))
1162 copy(lines, a.transcript)
1163 errStr := ""
1164 if a.lastErr != nil {
1165 errStr = a.lastErr.Error()
1166 }
1167 status := a.status
1168 activity := a.activity
1169 lastAssistant := a.lastAssistant
1170 outputTruncated := a.outputTruncated || a.streamingAssistantTruncated
1171 finished := a.finished
1172 a.mu.Unlock()
1173
1174 a.lifecycleMu.Lock()
1175 processState := a.processState
1176 turnState := a.turnState
1177 currentTurnID := a.currentTurnID
1178 lifetimeTurns := a.LifetimeTurns
1179 currentRunTurns := a.CurrentRunTurns
1180 attempt := a.Attempt
1181 lastActivity := a.lastActivity
1182 updatedAt := a.updatedAt
1183 result := cloneTurnResult(a.result)
1184 resultRef := a.resultRef
1185 requirement := a.visibleRequirementLocked()
1186 patchRef := a.patchRef
1187 changedFiles := append([]string(nil), a.changedFiles...)
1188 a.lifecycleMu.Unlock()
1189 return AgentSnapshot{
1190 ID: a.ID, Task: a.Task, Dir: a.Dir,
1191 Status: status, ProcessState: processState, TurnState: turnState,
1192 CurrentTurnID: currentTurnID, LifetimeTurns: lifetimeTurns, CurrentRunTurns: currentRunTurns, Attempt: attempt, Activity: activity,
1193 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1194 Err: errStr, Tail: tail, Lines: lines,
1195 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1196 Requirement: requirement, PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1197 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1198 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1199 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1200 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1201 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1202 }
1203}
1204
1205// SnapshotAll returns the supervisor's established status view: live agents
1206// are scoped to the active session, while durable terminal entries remain
1207// discoverable for recovery callers.
1208func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1209 f.mu.Lock()
1210 active := f.activeSession
1211 f.mu.Unlock()
1212 return f.snapshotMatching(func(a *Agent) bool {
1213 if active == "" || a.SessionID == "" || a.SessionID == active {
1214 return true
1215 }
1216 status := a.Status()
1217 return status != StatusRunning && status != StatusPending
1218 })
1219}
1220
1221// SnapshotAllSessions returns every agent, regardless of host session. It is
1222// the explicit cross-session view used by the dashboard's All filter.
1223func (f *Supervisor) SnapshotAllSessions() []AgentSnapshot {
1224 return f.snapshotMatching(func(*Agent) bool { return true })
1225}
1226
1227// SnapshotCurrentSession returns only agents owned by the active host
1228// session. When no active session is set, it includes every agent for
1229// command-line and test callers that do not establish a session boundary.
1230func (f *Supervisor) SnapshotCurrentSession() []AgentSnapshot {
1231 f.mu.Lock()
1232 active := f.activeSession
1233 f.mu.Unlock()
1234 return f.snapshotMatching(func(a *Agent) bool {
1235 return active == "" || a.SessionID == active
1236 })
1237}
1238
1239func (f *Supervisor) snapshotMatching(include func(*Agent) bool) []AgentSnapshot {
1240 agents := f.List()
1241 out := make([]AgentSnapshot, 0, len(agents))
1242 for _, a := range agents {
1243 if include(a) {
1244 out = append(out, a.Snapshot())
1245 }
1246 }
1247 // Sort by start time for a stable, deterministic listing.
1248 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1249 return out
1250}
1251
1252// agentSink is the Sink the Supervisor hands to each Runner.
1253type agentSink struct{ a *Agent }
1254
1255func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1256func (s agentSink) Transcript(chunk string) {
1257 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") ||
1258 strings.HasPrefix(chunk, "tool: ") || strings.HasPrefix(chunk, "tool result: ") {
1259 s.a.appendTranscript(chunk)
1260 return
1261 }
1262 s.a.appendAssistantMessage(chunk)
1263}
1264func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1265func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1266func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
1267func (s agentSink) resetStreamingAssistant() { s.a.resetStreamingAssistant() }
1268
1269func truncate(s string, n int) string {
1270 runes := []rune(s)
1271 if len(runes) <= n {
1272 return s
1273 }
1274 if n <= 3 {
1275 return strings.Repeat(".", n)
1276 }
1277 return string(runes[:n-3]) + "..."
1278}
1279
1280func lastN(lines []string, n int) []string {
1281 if len(lines) <= n {
1282 return lines
1283 }
1284 return lines[len(lines)-n:]
1285}