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