1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sort"
8 "strconv"
9 "strings"
10 "time"
11)
12
13// RequirementState is the durable outcome of work that the parent declared
14// mandatory before it may finish its own turn.
15type RequirementState string
16
17const (
18 RequirementPending RequirementState = "pending"
19 RequirementSatisfied RequirementState = "satisfied"
20 RequirementFailed RequirementState = "failed"
21 RequirementTimedOut RequirementState = "timed_out"
22 RequirementCanceled RequirementState = "canceled"
23 RequirementIndeterminate RequirementState = "indeterminate"
24)
25
26// RequirementSnapshot is persisted with its worker. A requirement targets a
27// delegated message turn rather than a process attempt because resumable
28// workers may execute several turns in one process.
29type RequirementSnapshot struct {
30 Required bool `json:"required"`
31 State RequirementState `json:"state,omitempty"`
32 TargetTurn int `json:"target_turn,omitempty"`
33 UpdatedAt time.Time `json:"updated_at,omitempty"`
34 ErrorCode string `json:"error_code,omitempty"`
35 Notified bool `json:"notified,omitempty"`
36}
37
38// Unmet reports whether this requirement still prevents parent completion.
39func (r RequirementSnapshot) Unmet() bool {
40 if !r.Required {
41 return false
42 }
43 return r.State != RequirementSatisfied
44}
45
46func (r RequirementSnapshot) pending() bool {
47 return r.Required && r.State == RequirementPending
48}
49
50func newRequirement(required bool, targetTurn int, now time.Time) RequirementSnapshot {
51 if !required {
52 return RequirementSnapshot{}
53 }
54 return RequirementSnapshot{
55 Required: true,
56 State: RequirementPending,
57 TargetTurn: targetTurn,
58 UpdatedAt: now.UTC(),
59 }
60}
61
62func (a *Agent) requirementSnapshot() RequirementSnapshot {
63 if a == nil {
64 return RequirementSnapshot{}
65 }
66 a.lifecycleMu.Lock()
67 defer a.lifecycleMu.Unlock()
68 return a.visibleRequirementLocked()
69}
70
71func (a *Agent) prepareRequired(targetTurn int) RequirementSnapshot {
72 a.lifecycleMu.Lock()
73 defer a.lifecycleMu.Unlock()
74 a.initRequirementSignalLocked()
75 previous := a.requirement
76 if targetTurn <= 0 {
77 nextTurn := a.LifetimeTurns + 1
78 if previous.pending() && previous.TargetTurn >= nextTurn {
79 return previous
80 }
81 targetTurn = nextTurn
82 if previous.TargetTurn >= targetTurn {
83 targetTurn = previous.TargetTurn + 1
84 }
85 }
86 a.requirement = RequirementSnapshot{
87 Required: true,
88 State: RequirementPending,
89 TargetTurn: targetTurn,
90 UpdatedAt: time.Now().UTC(),
91 }
92 a.signalRequirementLocked()
93 return previous
94}
95
96func (a *Agent) restoreRequirement(previous RequirementSnapshot) {
97 a.lifecycleMu.Lock()
98 a.initRequirementSignalLocked()
99 a.requirement = previous
100 a.requirementPersisting = false
101 a.signalRequirementLocked()
102 a.lifecycleMu.Unlock()
103}
104
105func (a *Agent) markRequirementNotified() {
106 if a == nil {
107 return
108 }
109 a.lifecycleMu.Lock()
110 if !a.requirement.Required || a.requirement.Notified {
111 a.lifecycleMu.Unlock()
112 return
113 }
114 previous := a.requirement
115 a.requirement.Notified = true
116 a.requirement.UpdatedAt = time.Now().UTC()
117 notified := a.requirement
118 a.lifecycleMu.Unlock()
119 if a.persistFn != nil {
120 if err := a.persistFn(a); err != nil {
121 a.lifecycleMu.Lock()
122 if a.requirement == notified {
123 a.requirement = previous
124 }
125 a.lifecycleMu.Unlock()
126 a.recordPersistenceError(err)
127 return
128 }
129 }
130 a.lifecycleMu.Lock()
131 if a.requirement == notified {
132 a.signalRequirementLocked()
133 }
134 a.lifecycleMu.Unlock()
135}
136
137func (a *Agent) resolveRequirement(step int, result *TurnResult, errMsg string, force bool) RequirementSnapshot {
138 if a == nil {
139 return RequirementSnapshot{}
140 }
141 a.lifecycleMu.Lock()
142 a.initRequirementSignalLocked()
143 current := a.requirement
144 if !current.pending() || (!force && step != current.TargetTurn) {
145 a.lifecycleMu.Unlock()
146 return current
147 }
148 if result != nil && strings.TrimSpace(result.TurnID) != "" && !requirementAcceptsResult(current, result) {
149 result = nil
150 if strings.TrimSpace(errMsg) == "" {
151 errMsg = "required turn ended without a matching durable result"
152 }
153 }
154 if result != nil && a.persistFn != nil && !unobservedRequirementResult(result) {
155 durable, err := readTurnResult(a.stateDirectory(""))
156 if err != nil || durable.TurnID != result.TurnID || durable.AgentID != result.AgentID {
157 if err == nil {
158 err = fmt.Errorf("durable result identity does not match agent %s turn %s", result.AgentID, result.TurnID)
159 }
160 persistenceErr := fmt.Errorf("persist required result: %w", err)
161 a.requirement.State = RequirementIndeterminate
162 a.requirement.ErrorCode = "required_outcome_unobserved"
163 a.requirement.Notified = false
164 a.requirement.UpdatedAt = time.Now().UTC()
165 resolved := a.requirement
166 a.requirementPersisting = true
167 a.lifecycleMu.Unlock()
168 a.recordPersistenceError(persistenceErr)
169 if err := a.persistFn(a); err != nil {
170 a.recordPersistenceError(err)
171 }
172 a.lifecycleMu.Lock()
173 if a.requirement == resolved {
174 a.requirementPersisting = false
175 a.signalRequirementLocked()
176 }
177 a.lifecycleMu.Unlock()
178 return resolved
179 }
180 }
181 state, errorCode := classifyRequirementOutcome(result, errMsg)
182 a.requirement.State = state
183 a.requirement.ErrorCode = errorCode
184 a.requirement.Notified = false
185 a.requirement.UpdatedAt = time.Now().UTC()
186 resolved := a.requirement
187 if a.persistFn != nil {
188 a.requirementPersisting = true
189 }
190 a.lifecycleMu.Unlock()
191 if a.persistFn != nil {
192 if err := a.persistFn(a); err != nil {
193 a.lifecycleMu.Lock()
194 if a.requirement == resolved {
195 a.requirement = current
196 a.requirementPersisting = false
197 }
198 a.lifecycleMu.Unlock()
199 a.recordPersistenceError(err)
200 return current
201 }
202 }
203 a.lifecycleMu.Lock()
204 if a.requirement == resolved {
205 a.requirementPersisting = false
206 a.signalRequirementLocked()
207 }
208 a.lifecycleMu.Unlock()
209 return resolved
210}
211
212func unobservedRequirementResult(result *TurnResult) bool {
213 return result != nil && result.Error != nil && result.Error.Code == "required_outcome_unobserved"
214}
215
216func requirementAcceptsResult(requirement RequirementSnapshot, result *TurnResult) bool {
217 if !requirement.pending() || result == nil {
218 return false
219 }
220 turnID := strings.TrimSpace(result.TurnID)
221 if strings.HasPrefix(turnID, "turn-") {
222 if turn, err := strconv.Atoi(strings.TrimPrefix(turnID, "turn-")); err == nil {
223 return turn == requirement.TargetTurn
224 }
225 }
226 // Legacy first-turn results did not always carry a turn id. They can
227 // satisfy only the initial requirement; retries fail closed without proof.
228 return requirement.TargetTurn == 1
229}
230
231func classifyRequirementOutcome(result *TurnResult, errMsg string) (RequirementState, string) {
232 if result != nil {
233 errorCode := ""
234 if result.Error != nil {
235 errorCode = strings.TrimSpace(result.Error.Code)
236 }
237 switch {
238 case errorCode == "required_outcome_unobserved":
239 return RequirementIndeterminate, "required_outcome_unobserved"
240 case result.ShutdownOrigin == ShutdownOriginDeadline || errorCode == "deadline_exceeded" || errorCode == "timeout":
241 return RequirementTimedOut, "deadline_exceeded"
242 case result.Status == ResultCanceled || errorCode == "canceled" || errorCode == "cancelled":
243 return RequirementCanceled, "canceled"
244 case result.Status == ResultFailed:
245 return RequirementFailed, "failed"
246 default:
247 return RequirementSatisfied, ""
248 }
249 }
250 lower := strings.ToLower(strings.TrimSpace(errMsg))
251 switch {
252 case lower == "":
253 return RequirementSatisfied, ""
254 case strings.Contains(lower, "deadline exceeded") || strings.Contains(lower, "timed out"):
255 return RequirementTimedOut, "deadline_exceeded"
256 case strings.Contains(lower, "canceled") || strings.Contains(lower, "cancelled"):
257 return RequirementCanceled, "canceled"
258 default:
259 return RequirementFailed, "failed"
260 }
261}
262
263func (a *Agent) visibleRequirementLocked() RequirementSnapshot {
264 requirement := a.requirement
265 if a.requirementPersisting {
266 requirement.State = RequirementPending
267 requirement.ErrorCode = ""
268 requirement.Notified = false
269 }
270 return requirement
271}
272
273func (a *Agent) initRequirementSignalLocked() {
274 if a.requirementChanged == nil {
275 a.requirementChanged = make(chan struct{})
276 }
277}
278
279func (a *Agent) signalRequirementLocked() {
280 a.initRequirementSignalLocked()
281 close(a.requirementChanged)
282 a.requirementChanged = make(chan struct{})
283}
284
285func (a *Agent) waitRequirement(ctx context.Context) (RequirementSnapshot, error) {
286 if ctx == nil {
287 ctx = context.Background()
288 }
289 for {
290 a.lifecycleMu.Lock()
291 a.initRequirementSignalLocked()
292 requirement := a.visibleRequirementLocked()
293 changed := a.requirementChanged
294 a.lifecycleMu.Unlock()
295 if !requirement.pending() {
296 return requirement, nil
297 }
298 select {
299 case <-ctx.Done():
300 return RequirementSnapshot{}, ctx.Err()
301 case <-changed:
302 }
303 }
304}
305
306var errRequiredAgentMissing = errors.New("subagents: required agent no longer exists")
307
308// WaitRequired blocks on requirement state transitions, never filesystem or
309// status polling, until the selected required turn reaches a terminal outcome.
310func (f *Supervisor) WaitRequired(ctx context.Context, id string) (RequirementSnapshot, error) {
311 a := f.Get(strings.TrimSpace(id))
312 if a == nil {
313 return RequirementSnapshot{}, fmt.Errorf("%w: %q", errRequiredAgentMissing, id)
314 }
315 requirement := a.requirementSnapshot()
316 if !requirement.Required {
317 return requirement, fmt.Errorf("subagents: agent %s is not required", a.ID)
318 }
319 return a.waitRequirement(ctx)
320}
321
322// MarkRequirementNotified records that the parent received the terminal
323// outcome through a blocking manager tool or host-injected update.
324func (f *Supervisor) MarkRequirementNotified(id string) error {
325 a := f.Get(strings.TrimSpace(id))
326 if a == nil {
327 return fmt.Errorf("subagents: no such agent %q", id)
328 }
329 a.markRequirementNotified()
330 return nil
331}
332
333// RequiredSnapshots returns required workers owned by the active host session.
334// Unlike dashboard snapshots, terminal workers from other sessions are never
335// included because they must not gate an unrelated parent conversation.
336func (f *Supervisor) RequiredSnapshots() []AgentSnapshot {
337 f.mu.Lock()
338 active := f.activeSession
339 agents := make([]*Agent, 0, len(f.order))
340 for _, id := range f.order {
341 if a := f.agents[id]; a != nil {
342 agents = append(agents, a)
343 }
344 }
345 f.mu.Unlock()
346
347 out := make([]AgentSnapshot, 0, len(agents))
348 for _, a := range agents {
349 if active != "" && a.SessionID != "" && a.SessionID != active {
350 continue
351 }
352 snapshot := a.Snapshot()
353 if snapshot.Requirement.Required {
354 out = append(out, snapshot)
355 }
356 }
357 sort.SliceStable(out, func(i, j int) bool { return out[i].Started.Before(out[j].Started) })
358 return out
359}
360
361// UnmetRequirements returns every required worker that still blocks parent
362// completion in the active session.
363func (f *Supervisor) UnmetRequirements() []AgentSnapshot {
364 required := f.RequiredSnapshots()
365 out := required[:0]
366 for _, snapshot := range required {
367 if snapshot.Requirement.Unmet() {
368 out = append(out, snapshot)
369 }
370 }
371 return out
372}
373
374// WaitActiveRequirements waits for all currently pending required workers in
375// the active session. It returns all required snapshots so the host can inject
376// both successful outcomes and unresolved failures before the next model turn.
377func (f *Supervisor) WaitActiveRequirements(ctx context.Context) ([]AgentSnapshot, error) {
378 if ctx == nil {
379 ctx = context.Background()
380 }
381 for {
382 pending := false
383 for _, snapshot := range f.RequiredSnapshots() {
384 if snapshot.Requirement.pending() {
385 pending = true
386 if _, err := f.WaitRequired(ctx, snapshot.ID); err != nil {
387 if errors.Is(err, errRequiredAgentMissing) {
388 break
389 }
390 return nil, err
391 }
392 break
393 }
394 }
395 if !pending {
396 return f.RequiredSnapshots(), nil
397 }
398 }
399}