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