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// A positive timeout limits the worker lifetime; otherwise it is cancellable
184// only when the supervisor shuts down or the agent is stopped.
185func (f *Supervisor) workerContext(timeout time.Duration) (context.Context, context.CancelFunc) {
186 base := f.lifetimeCtx
187 if base == nil {
188 base = context.Background()
189 }
190 if timeout > 0 {
191 return context.WithTimeout(base, timeout)
192 }
193 return context.WithCancel(base)
194}
195
196// SetRepoRoot updates the shared working directory for subsequently
197// spawned agents. Existing agents keep the directory they were started
198// with; callers use this when the host session changes cwd.
199func (f *Supervisor) SetRepoRoot(root string) {
200 f.mu.Lock()
201 defer f.mu.Unlock()
202 f.cfg.RepoRoot = root
203}
204
205// RepoRoot returns the working directory used by subsequently spawned
206// agents.
207func (f *Supervisor) RepoRoot() string {
208 f.mu.Lock()
209 defer f.mu.Unlock()
210 return f.cfg.RepoRoot
211}
212
213// MaxTurns returns the maximum prompt-level turns allowed for subsequently
214// spawned agents.
215func (f *Supervisor) MaxTurns() int {
216 f.mu.Lock()
217 defer f.mu.Unlock()
218 return f.cfg.Policy.MaxTurns
219}
220
221// SetFastMode updates the fast-mode setting for subsequently spawned
222// children. Existing agents keep the setting captured at their spawn
223// boundary so resumes remain deterministic.
224func (f *Supervisor) SetFastMode(enabled bool) {
225 f.mu.Lock()
226 f.cfg.FastMode = enabled
227 f.mu.Unlock()
228}
229
230// FastMode reports the setting used for subsequently spawned children.
231func (f *Supervisor) FastMode() bool {
232 f.mu.Lock()
233 defer f.mu.Unlock()
234 return f.cfg.FastMode
235}
236
237// SetWebSearchPolicy updates the parent capability used for subsequently
238// spawned generic children. Existing agents retain their persisted decision.
239func (f *Supervisor) SetWebSearchPolicy(policy WebSearchPolicy) {
240 f.mu.Lock()
241 f.cfg.WebSearchPolicy = policy
242 f.mu.Unlock()
243}
244
245// WebSearchPolicy reports the parent capability used for new generic
246// children.
247func (f *Supervisor) WebSearchPolicy() WebSearchPolicy {
248 f.mu.Lock()
249 defer f.mu.Unlock()
250 return f.cfg.WebSearchPolicy
251}
252
253// SetProvider updates the provider used by subsequently spawned children.
254// Existing agents keep their persisted provider selection.
255func (f *Supervisor) SetProvider(provider string) {
256 f.mu.Lock()
257 f.cfg.Provider = strings.TrimSpace(provider)
258 f.mu.Unlock()
259}
260
261// SetProviderSettings updates the effective connection settings used by
262// subsequently spawned children. Existing agents keep their persisted values
263// so a model/provider switch cannot silently change a running or resumed
264// worker's endpoint.
265func (f *Supervisor) SetProviderSettings(baseURL string, insecureTLS bool) {
266 f.mu.Lock()
267 f.cfg.BaseURL = strings.TrimSpace(baseURL)
268 f.cfg.InsecureTLS = insecureTLS
269 f.mu.Unlock()
270}
271
272// ProviderSettings returns the settings used by subsequently spawned
273// children. It is primarily useful to host integrations that need to verify
274// the manager follows the active provider selection.
275func (f *Supervisor) ProviderSettings() (string, bool) {
276 f.mu.Lock()
277 defer f.mu.Unlock()
278 return f.cfg.BaseURL, f.cfg.InsecureTLS
279}
280
281// SetActiveSession scopes the dashboard view (and Spawn stamping)
282// to a particular host zut session id. Pass empty to clear the
283// scope and revert to "show every agent" (the original behaviour).
284//
285// Existing in-memory agents keep their SessionID; only the filter
286// applied at snapshot time changes. So swapping the active session
287// with /sessions instantly re-narrows the dashboard without
288// touching any agent state.
289func (f *Supervisor) SetActiveSession(id string) {
290 f.mu.Lock()
291 defer f.mu.Unlock()
292 f.activeSession = id
293}
294
295// ActiveSession returns the current scope, mostly for tests and
296// diagnostics. Empty means "no scope; show everything".
297func (f *Supervisor) ActiveSession() string {
298 f.mu.Lock()
299 defer f.mu.Unlock()
300 return f.activeSession
301}
302
303// agentStateDir is the per-agent state directory laid out as:
304//
305// <root>/agents/<id>/
306// events.jsonl durable event log (runner-owned)
307// session.json persistent agent session (child-owned)
308// meta.json static metadata (id, task)
309//
310// The transient unix socket inbox lives in a probed runtime directory,
311// which may be on a different filesystem from this durable state.
312func (f *Supervisor) agentStateDir(id string) string {
313 return filepath.Join(f.cfg.Root, "agents", id)
314}
315
316func (f *Supervisor) uniqueAgentIDLocked(base string) string {
317 id := base
318 for suffix := 2; ; suffix++ {
319 _, inMemory := f.agents[id]
320 _, onDiskErr := os.Stat(f.agentStateDir(id))
321 if !inMemory && onDiskErr != nil {
322 return id
323 }
324 id = fmt.Sprintf("%s-%d", base, suffix)
325 }
326}
327
328// SpawnRequest configures a Spawn. Only Task is required; the rest
329// are optional. Model + Provider, when set, get baked into the
330// child argv as --model / --provider so the agent runs against the
331// chosen model regardless of the parent's current selection. A named
332// Subagent is passed to the child as --subagent, where it resolves the
333// markdown profile and applies its instructions. Reasoning is passed
334// as --reasoning when set. FastMode overrides the host setting when a
335// selected profile declares a fastMode preference.
336type SpawnRequest struct {
337 Task string
338 Model string // optional override; child resolves default if empty
339 Provider string // optional override; usually paired with Model
340 BaseURL string // optional provider endpoint override
341 // InsecureTLS enables the child to skip TLS verification for its
342 // effective custom provider endpoint.
343 InsecureTLS bool
344 Reasoning string // optional reasoning/thinking level override
345 // FastMode is an optional profile override. A non-nil value enables or
346 // disables fast mode for this child; nil inherits Config.FastMode.
347 FastMode *bool
348 Subagent string // optional named markdown profile
349
350 // WebSearchPolicy is an internal child capability override. Generic
351 // requests normally inherit the supervisor's parent decision; named
352 // requests use Allow only when their profile explicitly lists web_search.
353 WebSearchPolicy WebSearchPolicy
354
355 // ParentID is metadata used for per-parent scheduling limits. A
356 // non-empty RequesterAgentID identifies an untrusted child request;
357 // recursive spawning is rejected in v1.
358 ParentID string
359 BatchID string
360 RootSessionID string
361 RequesterAgentID string
362
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(timeout)
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 stateDir: stateDir,
538 lease: lease,
539 maxOutputBytes: f.cfg.Policy.MaxOutputBytes,
540 maxOutputLines: f.cfg.Policy.MaxOutputLines,
541 done: make(chan struct{}),
542 turnResults: make(chan *TurnResult, 16),
543 }
544 a.fastModeOverridesHost = fastModeOverridesHost
545 if a.RootSessionID == "" {
546 a.RootSessionID = sessionID
547 }
548 a.ctx, a.cancel = runCtx, cancel
549 a.persistFn = f.persistAgent
550 a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
551 a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
552 a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
553 a.runner = f.cfg.NewRunner(a)
554 if err := writeAgentMeta(stateDir, a); err != nil {
555 if a.cancel != nil {
556 a.cancel()
557 }
558 _ = workspace.Cleanup(context.Background())
559 _ = lease.Close()
560 leaseOwned = false
561 _ = os.RemoveAll(stateDir)
562 return nil, fmt.Errorf("subagents initial metadata: %w", err)
563 }
564
565 f.mu.Lock()
566 f.agents[id] = a
567 f.order = append(f.order, id)
568 f.queue = append(f.queue, a)
569 f.mu.Unlock()
570 leaseOwned = false
571
572 // Persisted metadata was written before queue admission, so a later
573 // supervisor can reconstruct work that never reached a process.
574 f.armQueueTimeout(a)
575 f.schedule()
576 return a, nil
577}
578
579// SendUserTurn is sugar for the common "send the next user turn"
580// case. It quotes nothing and forwards verbatim; callers are
581// expected to have already trimmed and expanded the text.
582// CancelTurn requests cancellation of the current turn while keeping the
583// worker process alive for a later follow-up.
584func (f *Supervisor) CancelTurn(id string) error {
585 a := f.Get(id)
586 if a == nil {
587 return fmt.Errorf("subagents: no such agent %q", id)
588 }
589 if a.inbox == nil {
590 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
591 }
592 a.setTurnState(TurnCanceling, a.CurrentTurnID())
593 return a.inbox.SendCommand(NewCommand(CommandTurnCancel, a.ID, a.CurrentTurnID(), TurnCancelPayload{Reason: "user"}))
594}
595
596func (f *Supervisor) SendUserTurn(id, text string) error {
597 a := f.Get(id)
598 if a == nil {
599 return fmt.Errorf("subagents: no such agent %q", id)
600 }
601 if a.inbox == nil {
602 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
603 }
604 return a.inbox.SendCommand(NewCommand(CommandTurnStart, a.ID, a.CurrentTurnID(), TurnStartPayload{Prompt: text}))
605}
606
607func (f *Supervisor) run(a *Agent) {
608 a.mu.Lock()
609 killedBeforeStart := a.status == StatusKilled
610 ctxErr := error(nil)
611 if a.ctx != nil {
612 ctxErr = a.ctx.Err()
613 }
614 a.mu.Unlock()
615 var err error
616 if killedBeforeStart || ctxErr != nil || !atomic.CompareAndSwapInt32(&a.launchState, 0, 1) {
617 if ctxErr != nil {
618 err = ctxErr
619 } else {
620 err = context.Canceled
621 }
622 } else {
623 a.setActivity("starting")
624 if a.runner == nil {
625 err = errors.New("subagents: agent has no runner")
626 } else {
627 err = a.runner.Run(a.ctx, agentSink{a: a})
628 }
629 }
630
631 a.mu.Lock()
632 a.finished = f.cfg.Now()
633 switch {
634 case a.status == StatusKilled:
635 // Stop requested a terminal state; preserve it.
636 case errors.Is(err, context.Canceled):
637 a.status = StatusKilled
638 a.activity = "cancelled"
639 a.lastErr = err
640 case errors.Is(err, context.DeadlineExceeded):
641 a.status = StatusFailed
642 a.activity = "timeout"
643 a.lastErr = err
644 case err != nil:
645 a.status = StatusFailed
646 a.activity = "error: " + truncate(err.Error(), 120)
647 a.lastErr = err
648 default:
649 a.status = StatusDone
650 a.activity = "done"
651 }
652 finalStatus := a.status
653 a.mu.Unlock()
654
655 switch finalStatus {
656 case StatusKilled:
657 a.setProcessState(ProcessKilled)
658 a.setTurnState(TurnCanceled, a.CurrentTurnID())
659 case StatusFailed:
660 a.setProcessState(ProcessExited)
661 a.setTurnState(TurnFailed, a.CurrentTurnID())
662 default:
663 a.setProcessState(ProcessExited)
664 if a.TurnState() == TurnRunning || a.TurnState() == TurnIdle {
665 a.setTurnState(TurnSucceeded, a.CurrentTurnID())
666 }
667 }
668
669 // The process slot is no longer occupied once Runner.Run has returned.
670 // Release it before potentially slow artifact capture and workspace
671 // cleanup so queued work can start without waiting on git/filesystem I/O.
672 f.releaseCapacity(a)
673
674 // Capture isolated output and write the structured result before the
675 // temporary workspace is removed. This ordering is the recovery
676 // invariant: a user can always inspect a durable result after failure.
677 f.captureWorkspace(a)
678 f.ensureResult(a, finalStatus, err)
679 if a.cancel != nil {
680 a.cancel()
681 }
682 if a.inbox != nil {
683 _ = a.inbox.Close()
684 }
685 a.setProcessPID(0)
686 f.persistAgent(a)
687 if a.workspaceCleanup != nil {
688 _ = a.workspaceCleanup()
689 }
690 _ = a.releaseLease()
691 a.closeDone()
692}
693
694// List returns a snapshot of every agent in creation order. The
695// returned slice is a copy; callers may iterate without holding the
696// subagent lock. Agent fields are read under their own mutex during
697// formatting in Snapshot.
698func (f *Supervisor) List() []*Agent {
699 f.mu.Lock()
700 defer f.mu.Unlock()
701 out := make([]*Agent, 0, len(f.order))
702 for _, id := range f.order {
703 out = append(out, f.agents[id])
704 }
705 return out
706}
707
708// Get returns the agent with the given (possibly truncated) id, or
709// nil. Matching is prefix-based so the user can type the first few
710// characters of a long id.
711func (f *Supervisor) Get(id string) *Agent {
712 f.mu.Lock()
713 defer f.mu.Unlock()
714 if a, ok := f.agents[id]; ok {
715 return a
716 }
717 // Prefix match.
718 var hits []*Agent
719 for k, a := range f.agents {
720 if strings.HasPrefix(k, id) {
721 hits = append(hits, a)
722 }
723 }
724 if len(hits) == 1 {
725 return hits[0]
726 }
727 return nil
728}
729
730var errDetachedStopTimeout = errors.New("detached worker stop timeout")
731
732const detachedStopPollInterval = 20 * time.Millisecond
733
734// waitForDetachedWorker waits for the worker inbox to disappear, returning a
735// sentinel when the grace period expires. Polling is interruptible so a
736// caller's context is not made to wait for the next fixed sleep interval.
737func waitForDetachedWorker(ctx context.Context, path string, timeout time.Duration) error {
738 if ctx == nil {
739 ctx = context.Background()
740 }
741 timer := time.NewTimer(timeout)
742 defer timer.Stop()
743 ticker := time.NewTicker(detachedStopPollInterval)
744 defer ticker.Stop()
745 for {
746 if err := ctx.Err(); err != nil {
747 return err
748 }
749 if !inboxLive(path) {
750 return nil
751 }
752 select {
753 case <-ctx.Done():
754 return ctx.Err()
755 case <-timer.C:
756 if !inboxLive(path) {
757 return nil
758 }
759 return errDetachedStopTimeout
760 case <-ticker.C:
761 }
762 }
763}
764
765// Stop cancels the agent's context. The Runner should observe the
766// cancellation and return promptly; the goroutine then finalises the
767// agent in StatusKilled. Also closes the supervisor-side inbox handle
768// so any pending send retries fail fast instead of dialing a
769// socket about to be unlinked. Stop uses a background context; callers
770// that need cancellation while waiting for a detached worker should use
771// StopContext.
772//
773// Stop is a no-op for any agent that's not in a live runnable state
774// — Done / Failed / Killed (already finalised). Reloaded Detached agents
775// are probed first: if their worker survived the supervisor, Stop sends the
776// shutdown command through the durable inbox instead of pretending no
777// process exists.
778func (f *Supervisor) Stop(id string) error {
779 return f.stop(context.Background(), id)
780}
781
782// StopContext is Stop with a context that can interrupt detached-worker
783// shutdown and its graceful and force-stop waits. The context is not held
784// under the supervisor operation lock while those waits run.
785func (f *Supervisor) StopContext(ctx context.Context, id string) error {
786 return f.stop(ctx, id)
787}
788
789func (f *Supervisor) stop(ctx context.Context, id string) error {
790 if ctx == nil {
791 ctx = context.Background()
792 }
793 f.operationMu.Lock()
794 a := f.Get(id)
795 if a == nil {
796 f.operationMu.Unlock()
797 return fmt.Errorf("no such agent %q", id)
798 }
799
800 // Coordinate the status transition with schedule. Holding f.mu while
801 // checking and changing a.status closes the small window where Stop could
802 // observe pending, schedule could promote the same agent to running, and
803 // the pending cleanup below could then finalize an agent that was already
804 // admitted to the runner.
805 f.mu.Lock()
806 a.mu.Lock()
807 status := a.status
808 if status == StatusDone || status == StatusFailed || status == StatusKilled {
809 a.mu.Unlock()
810 f.mu.Unlock()
811 f.operationMu.Unlock()
812 return nil
813 }
814 if status == StatusDetached {
815 path := a.InboxPath
816 inbox := a.inbox
817 pid := a.ProcessPIDValue()
818 a.mu.Unlock()
819 f.mu.Unlock()
820 // The detached worker may take the full grace period to exit. Do not
821 // serialize unrelated supervisor operations for that entire wait.
822 f.operationMu.Unlock()
823 return f.stopDetached(ctx, a, path, inbox, pid)
824 }
825 a.status = StatusKilled
826 a.activity = "canceling"
827 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
828 if status == StatusPending {
829 for i, queued := range f.queue {
830 if queued == a {
831 f.queue = append(f.queue[:i], f.queue[i+1:]...)
832 break
833 }
834 }
835 }
836 a.mu.Unlock()
837 f.mu.Unlock()
838 f.operationMu.Unlock()
839 a.setTurnState(TurnCanceling, a.CurrentTurnID())
840
841 if status == StatusPending {
842 a.setProcessState(ProcessKilled)
843 a.setTurnState(TurnCanceled, a.CurrentTurnID())
844 f.captureWorkspace(a)
845 f.ensureResult(a, StatusKilled, context.Canceled)
846 if a.cancel != nil {
847 a.cancel()
848 }
849 f.persistAgent(a)
850 if a.workspaceCleanup != nil {
851 _ = a.workspaceCleanup()
852 }
853 _ = a.releaseLease()
854 a.closeDone()
855 f.schedule()
856 if err := ctx.Err(); err != nil {
857 return err
858 }
859 return nil
860 }
861
862 // Ask a live worker to shut down cleanly first. If the socket is not
863 // ready, fall back immediately; a forceful context cancellation is
864 // safer than leaving a supervisor entry running forever.
865 if err := ctx.Err(); err != nil {
866 if a.cancel != nil {
867 a.cancel()
868 }
869 return err
870 }
871 graceful := false
872 if a.inbox != nil {
873 err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))
874 if err == nil {
875 graceful = true
876 }
877 }
878 if !graceful && a.cancel != nil {
879 a.cancel()
880 }
881 if graceful {
882 grace := f.cfg.Policy.CancelGracePeriod
883 if grace <= 0 {
884 grace = 10 * time.Second
885 }
886 go func() {
887 timer := time.NewTimer(grace)
888 defer timer.Stop()
889 select {
890 case <-a.done:
891 case <-timer.C:
892 if a.cancel != nil {
893 a.cancel()
894 }
895 if a.inbox != nil {
896 _ = a.inbox.Close()
897 }
898 }
899 }()
900 }
901 if err := ctx.Err(); err != nil {
902 if a.cancel != nil {
903 a.cancel()
904 }
905 if a.inbox != nil {
906 _ = a.inbox.Close()
907 }
908 return err
909 }
910 return nil
911}
912
913func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {
914 if inbox == nil {
915 return nil
916 }
917 if err := ctx.Err(); err != nil {
918 return err
919 }
920 if !inboxLive(path) {
921 return nil
922 }
923 if err := ctx.Err(); err != nil {
924 return err
925 }
926 if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {
927 if ctxErr := ctx.Err(); ctxErr != nil {
928 return ctxErr
929 }
930 return err
931 }
932 a.setActivity("shutdown requested")
933 grace := f.cfg.Policy.CancelGracePeriod
934 if grace <= 0 {
935 grace = 10 * time.Second
936 }
937 waitErr := waitForDetachedWorker(ctx, path, grace)
938 if waitErr != nil && !errors.Is(waitErr, errDetachedStopTimeout) {
939 return waitErr
940 }
941 if waitErr == nil {
942 f.markDetachedKilled(a)
943 return nil
944 }
945 if err := ctx.Err(); err != nil {
946 return err
947 }
948 if pid <= 0 {
949 return fmt.Errorf("agent %s live worker did not stop within %s and has no recorded pid", a.ID, grace)
950 }
951 if killErr := forceKillProcess(pid); killErr != nil {
952 return fmt.Errorf("agent %s force-stop: %w", a.ID, killErr)
953 }
954 if waitErr := waitForDetachedWorker(ctx, path, time.Second); waitErr != nil {
955 if errors.Is(waitErr, errDetachedStopTimeout) {
956 return fmt.Errorf("agent %s live worker did not stop after force-stop", a.ID)
957 }
958 return waitErr
959 }
960 f.markDetachedKilled(a)
961 return nil
962}
963
964func (f *Supervisor) markDetachedKilled(a *Agent) {
965 f.operationMu.Lock()
966 defer f.operationMu.Unlock()
967 a.mu.Lock()
968 a.status = StatusKilled
969 a.activity = "killed"
970 a.mu.Unlock()
971 a.setProcessState(ProcessKilled)
972 a.setTurnState(TurnCanceled, a.CurrentTurnID())
973 a.setProcessPID(0)
974 f.ensureResult(a, StatusKilled, context.Canceled)
975 f.persistAgent(a)
976 a.closeDone()
977}
978
979// StopAll cancels every running agent. Used on shutdown.
980func (f *Supervisor) StopAll() {
981 _ = f.stopAll(context.Background())
982}
983
984// StopAllContext is the context-aware shutdown variant. It returns the
985// context error when cancellation interrupts a detached-worker wait.
986func (f *Supervisor) StopAllContext(ctx context.Context) error {
987 return f.stopAll(ctx)
988}
989
990func (f *Supervisor) stopAll(ctx context.Context) error {
991 if ctx == nil {
992 ctx = context.Background()
993 }
994 agents := f.List()
995 var firstErr error
996 for _, a := range agents {
997 if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {
998 firstErr = err
999 }
1000 }
1001 defer func() {
1002 if f.lifetimeCancel != nil {
1003 f.lifetimeCancel()
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// Remove tears down the per-agent state for a terminated agent. It
1020// is an error to remove an agent that's still running; call Stop
1021// first and wait for the status to settle. Detached agents
1022// (reloaded from disk) remove cleanly because they have no live
1023// runner racing for the same files.
1024//
1025// Shared-workspace agents never touch source files during Remove; worktree
1026// agents have already captured their output and removed the isolated checkout
1027// during finalization. Remove then deletes the agent's durable state directory
1028// under <root>/agents/<id>/.
1029func (f *Supervisor) Remove(id string) error {
1030 f.operationMu.Lock()
1031 defer f.operationMu.Unlock()
1032 a := f.Get(id)
1033 if a == nil {
1034 return fmt.Errorf("no such agent %q", id)
1035 }
1036 a.mu.Lock()
1037 st := a.status
1038 inboxPath := a.InboxPath
1039 a.mu.Unlock()
1040 if st == StatusRunning || st == StatusPending {
1041 return fmt.Errorf("agent %s still %s", a.ID, st)
1042 }
1043 if inboxLive(inboxPath) {
1044 return fmt.Errorf("agent %s still has a live worker; stop it first", a.ID)
1045 }
1046 select {
1047 case <-a.done:
1048 default:
1049 return fmt.Errorf("agent %s is still finalizing", a.ID)
1050 }
1051 lease, err := acquireAgentLease(a.stateDirectory(f.cfg.Root))
1052 if err != nil {
1053 return fmt.Errorf("agent %s state is owned by another supervisor: %w", a.ID, err)
1054 }
1055 _ = lease.Close()
1056 // Best-effort cleanup of the per-agent state directory
1057 // (meta.json, events.jsonl, session.json). Failing here would leave
1058 // the user with no recourse,
1059 // so swallow the error.
1060 _ = os.RemoveAll(a.stateDirectory(f.cfg.Root))
1061 f.mu.Lock()
1062 delete(f.agents, a.ID)
1063 for i, k := range f.order {
1064 if k == a.ID {
1065 f.order = append(f.order[:i], f.order[i+1:]...)
1066 break
1067 }
1068 }
1069 f.mu.Unlock()
1070 return nil
1071}
1072
1073// Snapshot returns a read-only view of one agent. Safe for the TUI
1074// goroutine to call repeatedly; never blocks on the Runner.
1075type AgentSnapshot struct {
1076 ID string
1077 Task string
1078 Dir string
1079 Status Status
1080 ProcessState ProcessState
1081 TurnState TurnState
1082 CurrentTurnID string
1083 LifetimeTurns int
1084 CurrentRunTurns int
1085 Attempt int
1086 Activity string
1087 Started time.Time
1088 Finished time.Time
1089 CreatedAt time.Time
1090 UpdatedAt time.Time
1091 LastActivity time.Time
1092 Err string
1093 Tail string // last few transcript lines, joined with "\n"
1094 Lines []string // full transcript (already capped by Agent.appendTranscript)
1095 LastAssistant string // complete final assistant message, without role formatting
1096 Result *TurnResult
1097 ResultRef string
1098 PatchRef string
1099 ChangedFiles []string
1100 OutputTruncated bool
1101
1102 // Model, Provider, Reasoning, and the provider connection settings
1103 // expose the per-agent configuration captured at Spawn time. Empty
1104 // values inherit the child's defaults. Subagent is the selected named
1105 // markdown profile. The dashboard surfaces these so the user can
1106 // confirm the child configuration.
1107 Model string
1108 Provider string
1109 BaseURL string
1110 InsecureTLS bool
1111 Reasoning string
1112 FastMode bool
1113 Subagent string
1114 WorkspaceMode WorkspaceMode
1115 WorkspacePath string
1116
1117 // Paths to the agent's durable state. Surface them in the
1118 // snapshot so the dashboard /subagents view can
1119 // read events.jsonl or resume the session without going back through the Agent.
1120 InboxPath string
1121 EventLogPath string
1122 SessionPath string
1123}
1124
1125// Snapshot copies the live agent state into a value the caller can
1126// inspect at leisure.
1127func (a *Agent) Snapshot() AgentSnapshot {
1128 a.mu.Lock()
1129 tail := strings.Join(lastN(a.transcript, 6), "\n")
1130 lines := make([]string, len(a.transcript))
1131 copy(lines, a.transcript)
1132 errStr := ""
1133 if a.lastErr != nil {
1134 errStr = a.lastErr.Error()
1135 }
1136 status := a.status
1137 activity := a.activity
1138 lastAssistant := a.lastAssistant
1139 outputTruncated := a.outputTruncated || a.streamingAssistantTruncated
1140 finished := a.finished
1141 a.mu.Unlock()
1142
1143 a.lifecycleMu.Lock()
1144 processState := a.processState
1145 turnState := a.turnState
1146 currentTurnID := a.currentTurnID
1147 lifetimeTurns := a.LifetimeTurns
1148 currentRunTurns := a.CurrentRunTurns
1149 attempt := a.Attempt
1150 lastActivity := a.lastActivity
1151 updatedAt := a.updatedAt
1152 result := cloneTurnResult(a.result)
1153 resultRef := a.resultRef
1154 patchRef := a.patchRef
1155 changedFiles := append([]string(nil), a.changedFiles...)
1156 a.lifecycleMu.Unlock()
1157 return AgentSnapshot{
1158 ID: a.ID, Task: a.Task, Dir: a.Dir,
1159 Status: status, ProcessState: processState, TurnState: turnState,
1160 CurrentTurnID: currentTurnID, LifetimeTurns: lifetimeTurns, CurrentRunTurns: currentRunTurns, Attempt: attempt, Activity: activity,
1161 Started: a.Started, Finished: finished, CreatedAt: a.Started, UpdatedAt: updatedAt, LastActivity: lastActivity,
1162 Err: errStr, Tail: tail, Lines: lines,
1163 LastAssistant: lastAssistant, Result: result, ResultRef: resultRef,
1164 PatchRef: patchRef, ChangedFiles: changedFiles, OutputTruncated: outputTruncated,
1165 Model: a.Model, Provider: a.Provider, BaseURL: a.BaseURL,
1166 InsecureTLS: a.InsecureTLS, Reasoning: a.Reasoning,
1167 FastMode: a.FastMode, Subagent: a.Subagent, WorkspaceMode: a.WorkspaceMode,
1168 WorkspacePath: a.WorkspacePath, InboxPath: a.InboxPath,
1169 EventLogPath: a.EventLogPath, SessionPath: a.SessionPath,
1170 }
1171}
1172
1173// SnapshotAll returns snapshots of every agent in creation order,
1174// scoped to the active session when one is set.
1175//
1176// Scoping rules:
1177// - activeSession == "": no filter, every agent is returned
1178// (historical behaviour; used by tests and scripted callers).
1179// - activeSession != "": include live agents whose SessionID matches
1180// activeSession, plus unscoped agents. Detached and terminal
1181// agents from another host session remain visible so restart/recovery
1182// does not hide durable work.
1183func (f *Supervisor) SnapshotAll() []AgentSnapshot {
1184 agents := f.List()
1185 f.mu.Lock()
1186 active := f.activeSession
1187 f.mu.Unlock()
1188
1189 out := make([]AgentSnapshot, 0, len(agents))
1190 for _, a := range agents {
1191 if active != "" && a.SessionID != "" && a.SessionID != active {
1192 // Keep detached/terminal entries visible after a normal host
1193 // restart so durable recovery remains discoverable. Hide only
1194 // live work belonging to another in-process host session.
1195 status := a.Status()
1196 if status == StatusRunning || status == StatusPending {
1197 continue
1198 }
1199 }
1200 out = append(out, a.Snapshot())
1201 }
1202 // Sort by start time for a stable, deterministic listing.
1203 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
1204 return out
1205}
1206
1207// agentSink is the Sink the Supervisor hands to each Runner.
1208type agentSink struct{ a *Agent }
1209
1210func (s agentSink) Activity(msg string) { s.a.setActivity(msg) }
1211func (s agentSink) Transcript(chunk string) {
1212 if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") {
1213 s.a.appendTranscript(chunk)
1214 return
1215 }
1216 s.a.appendAssistantMessage(chunk)
1217}
1218func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) }
1219func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) }
1220func (s agentSink) assistantDelta(text string) { s.a.appendAssistantDelta(text) }
1221func (s agentSink) resetStreamingAssistant() { s.a.resetStreamingAssistant() }
1222
1223func truncate(s string, n int) string {
1224 runes := []rune(s)
1225 if len(runes) <= n {
1226 return s
1227 }
1228 if n <= 3 {
1229 return strings.Repeat(".", n)
1230 }
1231 return string(runes[:n-3]) + "..."
1232}
1233
1234func lastN(lines []string, n int) []string {
1235 if len(lines) <= n {
1236 return lines
1237 }
1238 return lines[len(lines)-n:]
1239}