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