1package agent
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "strings"
9
10 "github.com/bnema/zut/packages/agent/subagents"
11 "github.com/bnema/zut/packages/core"
12 "github.com/bnema/zut/packages/provider"
13)
14
15const requiredWorkerContextHeader = "[required-subagents update]"
16
17// WireRequiredWorkerGate composes host-enforced requirement handling with any
18// extension hooks already installed on the parent. Parent turns and tools stay
19// available while required work runs; terminal outcomes are injected on the
20// next turn, and only a terminal assistant response is held while an obligation
21// remains unmet.
22func (rt *subagentRuntime) WireRequiredWorkerGate(parent *core.Agent) {
23 if rt == nil || rt.supervisor == nil || parent == nil {
24 return
25 }
26 previousBeforeTurn := parent.BeforeTurnContext
27 parent.BeforeTurnContext = func(ctx context.Context, step int) (bool, string, string) {
28 var inherited string
29 if previousBeforeTurn != nil {
30 allowed, reason, contextText := previousBeforeTurn(ctx, step)
31 if !allowed {
32 return false, reason, contextText
33 }
34 inherited = contextText
35 }
36 if err := rt.waitRequiredWorkerReady(ctx); err != nil {
37 return false, fmt.Sprintf("loading persisted subagents failed: %v; resolve the supervisor reload error before continuing", err), inherited
38 }
39 required := rt.supervisor.RequiredSnapshots()
40 update := rt.formatRequiredWorkerContext(required)
41 if update == "" {
42 return true, "", inherited
43 }
44 return true, "", joinRequiredContext(inherited, update)
45 }
46
47 previousBeforeTool := parent.BeforeToolExecute
48 parent.BeforeToolExecute = func(call provider.ToolCallBlock) (bool, string, json.RawMessage) {
49 if previousBeforeTool != nil {
50 return previousBeforeTool(call)
51 }
52 return true, "", nil
53 }
54
55 previousBeforeAssistant := parent.BeforeAssistantMessage
56 parent.BeforeAssistantMessage = func(text string) (bool, string, string) {
57 unmet := rt.supervisor.UnmetRequirements()
58 if len(unmet) != 0 {
59 // Worker completion is delivered through the coordinator-owned tracker.
60 // Do not queue a synthetic reminder here: it races that terminal wave
61 // and can wake the parent before every required result is available.
62 return false, formatRequiredWorkerBlock(unmet), ""
63 }
64 if previousBeforeAssistant != nil {
65 allowed, reason, replacement := previousBeforeAssistant(text)
66 if !allowed {
67 return false, reason, replacement
68 }
69 if err := rt.acknowledgeSatisfiedRequirements(); err != nil {
70 return false, err.Error(), ""
71 }
72 return true, reason, replacement
73 }
74 if err := rt.acknowledgeSatisfiedRequirements(); err != nil {
75 return false, err.Error(), ""
76 }
77 return true, "", ""
78 }
79}
80
81// SetRequiredWorkerReady makes the gate wait for an asynchronous supervisor
82// reload before it checks persisted obligations. A nil channel means no reload
83// is pending. The error callback is read only after ready closes.
84func (rt *subagentRuntime) SetRequiredWorkerReady(ready <-chan struct{}, result func() error) {
85 if rt == nil {
86 return
87 }
88 rt.requiredReadyMu.Lock()
89 rt.requiredReady = ready
90 rt.requiredReadyErr = result
91 rt.requiredReadyMu.Unlock()
92}
93
94func (rt *subagentRuntime) waitRequiredWorkerReady(ctx context.Context) error {
95 rt.requiredReadyMu.RLock()
96 ready := rt.requiredReady
97 result := rt.requiredReadyErr
98 rt.requiredReadyMu.RUnlock()
99 if ready == nil {
100 return nil
101 }
102 select {
103 case <-ctx.Done():
104 return ctx.Err()
105 case <-ready:
106 if result != nil {
107 return result()
108 }
109 return nil
110 }
111}
112
113func (rt *subagentRuntime) acknowledgeSatisfiedRequirements() error {
114 var result error
115 for _, snapshot := range rt.supervisor.RequiredSnapshots() {
116 if snapshot.Requirement.State == subagents.RequirementSatisfied && !snapshot.Requirement.Notified {
117 result = errors.Join(result, rt.supervisor.MarkRequirementNotified(snapshot.ID))
118 }
119 }
120 return result
121}
122
123const requiredWorkerContextRuneBudget = 6000
124
125func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {
126 rt.requiredContextMu.Lock()
127 defer rt.requiredContextMu.Unlock()
128 contextText := formatRequiredWorkerContextWithReported(required, rt.reportedRequiredUnmet)
129 if rt.reportedRequiredUnmet == nil {
130 rt.reportedRequiredUnmet = make(map[string]struct{})
131 }
132 for _, snapshot := range required {
133 if snapshot.Requirement.Unmet() {
134 rt.reportedRequiredUnmet[snapshot.ID] = struct{}{}
135 }
136 }
137 return contextText
138}
139
140func formatRequiredWorkerContextWithReported(required []subagents.AgentSnapshot, previouslyReported map[string]struct{}) string {
141 var items []subagents.AgentSnapshot
142 for _, snapshot := range required {
143 requirement := snapshot.Requirement
144 if !requirement.Required || requirement.State == subagents.RequirementPending {
145 continue
146 }
147 if requirement.Notified && !requirement.Unmet() {
148 continue
149 }
150 items = append(items, snapshot)
151 }
152 if len(items) == 0 {
153 return ""
154 }
155 var sb strings.Builder
156 sb.WriteString(requiredWorkerContextHeader)
157 sb.WriteString("\nThe host waited for required delegated work before this model turn. These outcomes are durable and authoritative:\n")
158 for _, snapshot := range items {
159 requirement := snapshot.Requirement
160 if requirement.Unmet() {
161 if _, reported := previouslyReported[snapshot.ID]; reported {
162 fmt.Fprintf(&sb, "\n- %s: %s (turn %d; previously reported; diagnostic: %s)\n", snapshot.ID, requirement.State, requirement.TargetTurn, subagents.ResultRef(snapshot.ID))
163 continue
164 }
165 }
166 fmt.Fprintf(&sb, "\n- %s: %s (turn %d)\n task: %s\n", snapshot.ID, requirement.State, requirement.TargetTurn, strings.TrimSpace(snapshot.Task))
167 if requirement.ErrorCode != "" {
168 fmt.Fprintf(&sb, " error_code: %s\n diagnostic: %s\n", requirement.ErrorCode, subagents.ResultRef(snapshot.ID))
169 }
170 output := strings.TrimSpace(snapshot.LastAssistant)
171 if snapshot.Result != nil && strings.TrimSpace(snapshot.Result.Output) != "" {
172 output = strings.TrimSpace(snapshot.Result.Output)
173 }
174 if output != "" {
175 fmt.Fprintf(&sb, " result:\n%s\n", indentRequiredOutput(output, " ", 6000))
176 }
177 }
178 if hasIndeterminateRequirement(items) {
179 sb.WriteString("\nA required outcome is indeterminate after host restart. Do not retry it automatically: the user must inspect the durable result or external side effect, then explicitly resume, restart, or remove the terminal worker.")
180 } else if hasUnmetRequirement(items) {
181 sb.WriteString("\nAt least one required outcome is unresolved. Retry that worker with subagent_resume; do not emit a terminal answer. Only the user can waive it by removing the terminal worker with /subagents rm.")
182 } else {
183 sb.WriteString("\nAll listed requirements are satisfied. Use their results before answering.")
184 }
185 return truncateRequiredContext(sb.String(), requiredWorkerContextRuneBudget)
186}
187
188func truncateRequiredContext(text string, limit int) string {
189 runes := []rune(text)
190 if len(runes) <= limit {
191 return text
192 }
193 return string(runes[:limit])
194}
195
196func formatRequiredWorkerBlock(unmet []subagents.AgentSnapshot) string {
197 var sb strings.Builder
198 sb.WriteString(requiredWorkerContextHeader)
199 sb.WriteString("\nTerminal completion is blocked by required delegated work:")
200 for _, snapshot := range unmet {
201 fmt.Fprintf(&sb, "\n- %s: %s — %s", snapshot.ID, snapshot.Requirement.State, strings.TrimSpace(snapshot.Task))
202 if snapshot.Requirement.ErrorCode != "" {
203 fmt.Fprintf(&sb, " (%s; diagnostic: %s)", snapshot.Requirement.ErrorCode, subagents.ResultRef(snapshot.ID))
204 }
205 }
206 if hasIndeterminateRequirement(unmet) {
207 sb.WriteString("\nDo not retry an indeterminate outcome automatically. The user must inspect its durable result or external side effect, then explicitly resume, restart, or remove the worker.")
208 } else {
209 sb.WriteString("\nRetry unresolved work with subagent_resume. Do not claim completion. The user may explicitly waive a terminal worker with /subagents rm.")
210 }
211 return sb.String()
212}
213
214func hasIndeterminateRequirement(required []subagents.AgentSnapshot) bool {
215 for _, snapshot := range required {
216 if snapshot.Requirement.State == subagents.RequirementIndeterminate {
217 return true
218 }
219 }
220 return false
221}
222
223func hasUnmetRequirement(required []subagents.AgentSnapshot) bool {
224 for _, snapshot := range required {
225 if snapshot.Requirement.Unmet() {
226 return true
227 }
228 }
229 return false
230}
231
232func joinRequiredContext(first, second string) string {
233 first = strings.TrimSpace(first)
234 second = strings.TrimSpace(second)
235 switch {
236 case first == "":
237 return second
238 case second == "":
239 return first
240 default:
241 return first + "\n\n" + second
242 }
243}
244
245func indentRequiredOutput(text, prefix string, maxRunes int) string {
246 text = strings.TrimSpace(text)
247 runes := []rune(text)
248 if maxRunes > 0 && len(runes) > maxRunes {
249 text = string(runes[:maxRunes]) + "\n…(truncated)"
250 }
251 return prefix + strings.ReplaceAll(text, "\n", "\n"+prefix)
252}