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