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