1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "path/filepath"
9 "strings"
10 "sync/atomic"
11 "time"
12)
13
14var errStartupTimeout = errors.New("subagents: worker startup timeout")
15
16func (f *Supervisor) resolveWebSearchPolicy(req SpawnRequest) WebSearchPolicy {
17 // Inherit is valid only as the request's unresolved value. Unknown values
18 // are not another way to inherit the parent's capability.
19 switch req.WebSearchPolicy {
20 case WebSearchInherit, WebSearchDeny, WebSearchAllow:
21 default:
22 return WebSearchDeny
23 }
24
25 f.mu.Lock()
26 parent := f.cfg.WebSearchPolicy
27 allowed := f.cfg.Policy.allowedTool("web_search")
28 f.mu.Unlock()
29
30 // A child can never exceed its parent capability or the supervisor policy.
31 if parent != WebSearchAllow || !allowed || req.WebSearchPolicy == WebSearchDeny {
32 return WebSearchDeny
33 }
34 if req.Subagent != "" {
35 // Named profiles need an additional explicit opt-in. Passing an allow
36 // value in SpawnRequest cannot substitute for the profile's tools list.
37 return NamedWebSearchPolicy(req.Tools)
38 }
39 return WebSearchAllow
40}
41
42func (f *Supervisor) validateSpawnRequest(req SpawnRequest) error {
43 if err := f.validateSpawnOptions(req); err != nil {
44 return err
45 }
46 mode := req.WorkspaceMode
47 if mode == "" {
48 mode = WorkspaceShared
49 }
50 if mode != WorkspaceShared && mode != WorkspaceWorktree {
51 return fmt.Errorf("subagents: unknown workspace mode %q", mode)
52 }
53 if req.WorkspaceCapture != "" && req.WorkspaceCapture != CapturePatch && req.WorkspaceCapture != CaptureDiff {
54 return fmt.Errorf("subagents: unknown workspace capture mode %q", req.WorkspaceCapture)
55 }
56
57 f.mu.Lock()
58 root := f.cfg.RepoRoot
59 allowedRoots := append([]string(nil), f.cfg.Policy.AllowedRoots...)
60 f.mu.Unlock()
61 return validateWorkspaceRoot(root, mode, allowedRoots)
62}
63
64func (f *Supervisor) validateSpawnOptions(req SpawnRequest) error {
65 p := f.cfg.Policy
66 if req.MaxTurns < 0 || req.MaxTurns > p.MaxTurns {
67 return fmt.Errorf("subagents: max_turns must be 0 through %d (zero uses the policy default)", p.MaxTurns)
68 }
69 if req.Timeout < 0 {
70 return errorsInvalid("timeout must not be negative")
71 }
72 for _, tool := range req.Tools {
73 name := strings.TrimSpace(tool)
74 if name == "" {
75 continue
76 }
77 if !p.allowedTool(name) {
78 return fmt.Errorf("subagents: tool %q is not allowed by policy", name)
79 }
80 }
81 return nil
82}
83
84func validateWorkspaceRoot(root string, mode WorkspaceMode, allowedRoots []string) error {
85 if root == "" {
86 return errorsInvalid("repository root is empty")
87 }
88 if len(allowedRoots) > 0 {
89 ok := false
90 for _, allowed := range allowedRoots {
91 if pathWithin(root, allowed) {
92 ok = true
93 break
94 }
95 }
96 if !ok {
97 return fmt.Errorf("subagents: repository root is outside allowed roots")
98 }
99 }
100 if mode == WorkspaceWorktree {
101 if _, err := os.Stat(filepath.Join(root, ".git")); err != nil {
102 // Worktrees can be attached to repositories with a .git file or
103 // a parent checkout. Let the workspace adapter provide the
104 // detailed Git error, but fail obvious non-checkouts here.
105 if _, statErr := os.Stat(filepath.Join(root, "HEAD")); statErr != nil {
106 return fmt.Errorf("subagents: worktree mode requires a Git checkout: %w", err)
107 }
108 }
109 }
110 return nil
111}
112
113func errorsInvalid(message string) error { return fmt.Errorf("subagents: %s", message) }
114
115func pathWithin(path, root string) bool {
116 pathAbs, err := containmentPath(path)
117 if err != nil {
118 return false
119 }
120 rootAbs, err := containmentPath(root)
121 if err != nil {
122 return false
123 }
124 pathAbs = filepath.Clean(pathAbs)
125 rootAbs = filepath.Clean(rootAbs)
126 rel, err := filepath.Rel(rootAbs, pathAbs)
127 if err != nil {
128 return false
129 }
130 return rel == "." || (rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)))
131}
132
133// containmentPath returns a canonical path even when the final path does not
134// exist yet. EvalSymlinks cannot resolve a missing event or session file, so
135// resolve the nearest existing ancestor and append the missing components.
136// Without this, a symlinked temporary directory can make a valid child path
137// look outside its state directory on macOS and Windows.
138func containmentPath(path string) (string, error) {
139 absolute, err := filepath.Abs(path)
140 if err != nil {
141 return "", err
142 }
143 absolute = filepath.Clean(absolute)
144
145 current := absolute
146 var missing []string
147 for {
148 if _, statErr := os.Lstat(current); statErr == nil {
149 evaluated, evalErr := filepath.EvalSymlinks(current)
150 if evalErr != nil {
151 return "", evalErr
152 }
153 for i := len(missing) - 1; i >= 0; i-- {
154 evaluated = filepath.Join(evaluated, missing[i])
155 }
156 return filepath.Clean(evaluated), nil
157 } else if !os.IsNotExist(statErr) {
158 return "", statErr
159 }
160
161 parent := filepath.Dir(current)
162 if parent == current {
163 return absolute, nil
164 }
165 missing = append(missing, filepath.Base(current))
166 current = parent
167 }
168}
169
170func (f *Supervisor) schedule() {
171 var starts []*Agent
172 f.mu.Lock()
173 for len(f.queue) > 0 && f.active < f.cfg.Policy.MaxConcurrent {
174 idx := f.nextRunnableIndexLocked()
175 if idx < 0 {
176 break
177 }
178 a := f.queue[idx]
179 f.queue = append(f.queue[:idx], f.queue[idx+1:]...)
180
181 a.mu.Lock()
182 if a.status != StatusPending {
183 a.mu.Unlock()
184 continue
185 }
186 a.status = StatusRunning
187 a.activity = "starting"
188 a.mu.Unlock()
189 a.setProcessState(ProcessStarting)
190 prompt, _ := a.ResumePromptInfo()
191 initialTurn := (!a.Resuming && strings.TrimSpace(a.Task) != "") || prompt != ""
192 if initialTurn {
193 a.setTurnState(TurnQueued, "")
194 } else {
195 a.setTurnState(TurnIdle, "")
196 }
197 a.incrementAttempt()
198
199 f.active++
200 key := parentKey(a)
201 f.activeByParent[key]++
202 if a.BatchID != "" {
203 f.activeByBatch[a.BatchID]++
204 }
205 starts = append(starts, a)
206 }
207 f.mu.Unlock()
208
209 for _, a := range starts {
210 f.persistAgent(a)
211 f.startLifecycleMonitor(a)
212 f.armStartupTimeout(a)
213 go f.run(a)
214 }
215}
216
217func (f *Supervisor) startLifecycleMonitor(a *Agent) {
218 if a == nil || f.cfg.Policy.IdleTimeout <= 0 {
219 return
220 }
221 interval := f.cfg.Policy.HeartbeatInterval / 2
222 if interval <= 0 || interval > time.Minute {
223 interval = time.Minute
224 }
225 go func() {
226 ticker := time.NewTicker(interval)
227 defer ticker.Stop()
228 heartbeatGrace := f.cfg.Policy.HeartbeatInterval + f.cfg.Policy.ReconnectTimeout
229 for {
230 select {
231 case <-a.done:
232 return
233 case <-ticker.C:
234 if a.ProcessState() != ProcessAlive {
235 continue
236 }
237 last := a.LastActivity()
238 if !last.IsZero() && heartbeatGrace > 0 && time.Since(last) >= heartbeatGrace {
239 if err := f.Ping(a.ID); err != nil {
240 a.setProcessState(ProcessDetached)
241 a.setActivity("detached")
242 f.persistAgent(a)
243 continue
244 }
245 }
246 if a.TurnState() == TurnIdle && !last.IsZero() && time.Since(last) >= f.cfg.Policy.IdleTimeout {
247 _ = f.Stop(a.ID)
248 return
249 }
250 }
251 }
252 }()
253}
254
255// Ping asks a live worker for a heartbeat. A detached or terminal worker
256// returns the same not-ready error as other inbox operations.
257func (f *Supervisor) Ping(id string) error {
258 a := f.Get(id)
259 if a == nil {
260 return fmt.Errorf("subagents: no such agent %q", id)
261 }
262 if a.inbox == nil {
263 return fmt.Errorf("subagents: agent %s has no inbox", a.ID)
264 }
265 return a.inbox.SendCommand(NewCommand(CommandAgentPing, a.ID, a.CurrentTurnID(), AgentPingPayload{}))
266}
267
268func (f *Supervisor) nextRunnableIndexLocked() int {
269 if f.active >= f.cfg.Policy.MaxConcurrent {
270 return -1
271 }
272 for i, a := range f.queue {
273 if a == nil {
274 continue
275 }
276 a.mu.Lock()
277 status := a.status
278 a.mu.Unlock()
279 if status != StatusPending {
280 continue
281 }
282 if f.cfg.Policy.MaxConcurrentPerParent > 0 && f.activeByParent[parentKey(a)] >= f.cfg.Policy.MaxConcurrentPerParent {
283 continue
284 }
285 if a.BatchID != "" {
286 if batch := f.batches[a.BatchID]; batch != nil && batch.MaxConcurrent > 0 && f.activeByBatch[a.BatchID] >= batch.MaxConcurrent {
287 continue
288 }
289 }
290 return i
291 }
292 return -1
293}
294
295func parentKey(a *Agent) string {
296 if a == nil || strings.TrimSpace(a.ParentID) == "" {
297 return "supervisor"
298 }
299 return a.ParentID
300}
301
302func (f *Supervisor) armQueueTimeout(a *Agent) {
303 if a == nil {
304 return
305 }
306 if d := f.cfg.Policy.QueueTimeout; d > 0 {
307 time.AfterFunc(d, func() { f.expireQueued(a.ID) })
308 }
309 go func() {
310 <-a.ctx.Done()
311 f.cancelQueued(a.ID, a.ctx.Err())
312 }()
313}
314
315func (f *Supervisor) expireQueued(id string) {
316 f.cancelQueued(id, fmt.Errorf("subagents: queue timeout after %s waiting for an execution slot", f.cfg.Policy.QueueTimeout))
317}
318
319func (f *Supervisor) armStartupTimeout(a *Agent) {
320 if a == nil {
321 return
322 }
323 d := f.cfg.Policy.StartupTimeout
324 if d <= 0 {
325 return
326 }
327 go func() {
328 timer := time.NewTimer(d)
329 defer timer.Stop()
330 select {
331 case <-a.done:
332 return
333 case <-timer.C:
334 f.expireStarting(a, d)
335 }
336 }()
337}
338
339func (f *Supervisor) expireStarting(a *Agent, d time.Duration) {
340 if a == nil {
341 return
342 }
343 cause := fmt.Errorf("%w after %s waiting for agent_ready; inspect the worker output and retry", errStartupTimeout, d)
344 if !a.markStartupTimeout(cause) {
345 return
346 }
347 f.persistAgent(a)
348 if a.cancel != nil {
349 a.cancel()
350 }
351}
352
353func (f *Supervisor) cancelQueued(id string, cause error) {
354 a := f.Get(id)
355 if a == nil {
356 return
357 }
358
359 // schedule and Stop use f.mu as the admission lock. Make queued
360 // cancellation take the same lock while it changes status and removes the
361 // entry, so a timeout/cancel cannot finalize an agent concurrently with
362 // scheduler admission.
363 f.mu.Lock()
364 a.mu.Lock()
365 if a.status != StatusPending {
366 a.mu.Unlock()
367 f.mu.Unlock()
368 return
369 }
370 a.status = StatusFailed
371 atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
372 if errors.Is(cause, context.Canceled) {
373 a.activity = "cancelled while queued"
374 } else {
375 a.activity = "queue timeout: no execution slot became available"
376 }
377 a.lastErr = cause
378 a.finished = f.cfg.Now()
379 for i, queued := range f.queue {
380 if queued == a {
381 f.queue = append(f.queue[:i], f.queue[i+1:]...)
382 break
383 }
384 }
385 a.mu.Unlock()
386 f.mu.Unlock()
387 a.setProcessState(ProcessExited)
388 if cause == context.Canceled {
389 a.setTurnState(TurnCanceled, "")
390 } else {
391 a.setTurnState(TurnFailed, "")
392 }
393 f.captureWorkspace(a)
394 f.ensureResult(a, StatusFailed, cause)
395 if a.cancel != nil {
396 a.cancel()
397 }
398 f.persistAgent(a)
399 if a.workspaceCleanup != nil {
400 _ = a.workspaceCleanup()
401 }
402 _ = a.releaseLease()
403 a.closeDone()
404 f.schedule()
405}
406
407func (f *Supervisor) releaseCapacity(a *Agent) {
408 f.mu.Lock()
409 if f.active > 0 {
410 f.active--
411 }
412 key := parentKey(a)
413 if f.activeByParent[key] > 0 {
414 f.activeByParent[key]--
415 }
416 if a != nil && a.BatchID != "" && f.activeByBatch[a.BatchID] > 0 {
417 f.activeByBatch[a.BatchID]--
418 }
419 f.mu.Unlock()
420 f.schedule()
421}
422
423func (f *Supervisor) persistAgent(a *Agent) error {
424 if a == nil {
425 return errors.New("subagents: nil agent")
426 }
427 stateDir := a.stateDirectory(f.cfg.Root)
428 if err := writeAgentMeta(stateDir, a); err != nil {
429 // Lifecycle callbacks cannot all return an error to their caller. Keep
430 // the failure visible in the agent snapshot instead of silently
431 // allowing a queued prompt or counter to disappear on restart.
432 a.recordPersistenceError(err)
433 return err
434 }
435 return nil
436}