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