1package agent
2
3import (
4 "context"
5 "encoding/json"
6 "fmt"
7 "sort"
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 reason := formatRequiredWorkerBlock(unmet)
60 if rt.noteRequiredBlock(unmet) {
61 parent.QueueMessage(reason)
62 }
63 return false, reason, ""
64 }
65 rt.clearRequiredBlock()
66 if previousBeforeAssistant != nil {
67 allowed, reason, replacement := previousBeforeAssistant(text)
68 if !allowed {
69 return false, reason, replacement
70 }
71 rt.acknowledgeSatisfiedRequirements()
72 return true, reason, replacement
73 }
74 rt.acknowledgeSatisfiedRequirements()
75 return true, "", ""
76 }
77}
78
79// SetRequiredWorkerReady makes the gate wait for an asynchronous supervisor
80// reload before it checks persisted obligations. A nil channel means no reload
81// is pending. The error callback is read only after ready closes.
82func (rt *subagentRuntime) SetRequiredWorkerReady(ready <-chan struct{}, result func() error) {
83 if rt == nil {
84 return
85 }
86 rt.requiredReadyMu.Lock()
87 rt.requiredReady = ready
88 rt.requiredReadyErr = result
89 rt.requiredReadyMu.Unlock()
90}
91
92func (rt *subagentRuntime) waitRequiredWorkerReady(ctx context.Context) error {
93 rt.requiredReadyMu.RLock()
94 ready := rt.requiredReady
95 result := rt.requiredReadyErr
96 rt.requiredReadyMu.RUnlock()
97 if ready == nil {
98 return nil
99 }
100 select {
101 case <-ctx.Done():
102 return ctx.Err()
103 case <-ready:
104 if result != nil {
105 return result()
106 }
107 return nil
108 }
109}
110
111func (rt *subagentRuntime) acknowledgeSatisfiedRequirements() {
112 for _, snapshot := range rt.supervisor.RequiredSnapshots() {
113 if snapshot.Requirement.State == subagents.RequirementSatisfied && !snapshot.Requirement.Notified {
114 _ = rt.supervisor.MarkRequirementNotified(snapshot.ID)
115 }
116 }
117}
118
119const requiredWorkerContextRuneBudget = 6000
120
121func (rt *subagentRuntime) noteRequiredBlock(unmet []subagents.AgentSnapshot) bool {
122 ids := make([]string, 0, len(unmet))
123 for _, snapshot := range unmet {
124 ids = append(ids, fmt.Sprintf("%s:%s:%d:%s", snapshot.ID, snapshot.Requirement.State, snapshot.Requirement.TargetTurn, snapshot.Requirement.ErrorCode))
125 }
126 sort.Strings(ids)
127 key := strings.Join(ids, "\n")
128 rt.requiredContextMu.Lock()
129 defer rt.requiredContextMu.Unlock()
130 if key == rt.lastRequiredBlock {
131 return false
132 }
133 rt.lastRequiredBlock = key
134 return true
135}
136
137func (rt *subagentRuntime) clearRequiredBlock() {
138 rt.requiredContextMu.Lock()
139 rt.lastRequiredBlock = ""
140 rt.requiredContextMu.Unlock()
141}
142
143func (rt *subagentRuntime) formatRequiredWorkerContext(required []subagents.AgentSnapshot) string {
144 rt.requiredContextMu.Lock()
145 defer rt.requiredContextMu.Unlock()
146 contextText := formatRequiredWorkerContextWithReported(required, rt.reportedRequiredUnmet)
147 if rt.reportedRequiredUnmet == nil {
148 rt.reportedRequiredUnmet = make(map[string]struct{})
149 }
150 for _, snapshot := range required {
151 if snapshot.Requirement.Unmet() {
152 rt.reportedRequiredUnmet[snapshot.ID] = struct{}{}
153 }
154 }
155 return contextText
156}
157
158func formatRequiredWorkerContextWithReported(required []subagents.AgentSnapshot, previouslyReported map[string]struct{}) string {
159 var items []subagents.AgentSnapshot
160 for _, snapshot := range required {
161 requirement := snapshot.Requirement
162 if !requirement.Required || requirement.State == subagents.RequirementPending {
163 continue
164 }
165 if requirement.Notified && !requirement.Unmet() {
166 continue
167 }
168 items = append(items, snapshot)
169 }
170 if len(items) == 0 {
171 return ""
172 }
173 var sb strings.Builder
174 sb.WriteString(requiredWorkerContextHeader)
175 sb.WriteString("\nThe host waited for required delegated work before this model turn. These outcomes are durable and authoritative:\n")
176 for _, snapshot := range items {
177 requirement := snapshot.Requirement
178 if requirement.Unmet() {
179 if _, reported := previouslyReported[snapshot.ID]; reported {
180 fmt.Fprintf(&sb, "\n- %s: %s (turn %d; previously reported; diagnostic: %s)\n", snapshot.ID, requirement.State, requirement.TargetTurn, subagents.ResultRef(snapshot.ID))
181 continue
182 }
183 }
184 fmt.Fprintf(&sb, "\n- %s: %s (turn %d)\n task: %s\n", snapshot.ID, requirement.State, requirement.TargetTurn, strings.TrimSpace(snapshot.Task))
185 if requirement.ErrorCode != "" {
186 fmt.Fprintf(&sb, " error_code: %s\n diagnostic: %s\n", requirement.ErrorCode, subagents.ResultRef(snapshot.ID))
187 }
188 output := strings.TrimSpace(snapshot.LastAssistant)
189 if snapshot.Result != nil && strings.TrimSpace(snapshot.Result.Output) != "" {
190 output = strings.TrimSpace(snapshot.Result.Output)
191 }
192 if output != "" {
193 fmt.Fprintf(&sb, " result:\n%s\n", indentRequiredOutput(output, " ", 6000))
194 }
195 }
196 if hasIndeterminateRequirement(items) {
197 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.")
198 } else if hasUnmetRequirement(items) {
199 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.")
200 } else {
201 sb.WriteString("\nAll listed requirements are satisfied. Use their results before answering.")
202 }
203 return truncateRequiredContext(sb.String(), requiredWorkerContextRuneBudget)
204}
205
206func truncateRequiredContext(text string, limit int) string {
207 runes := []rune(text)
208 if len(runes) <= limit {
209 return text
210 }
211 return string(runes[:limit])
212}
213
214func formatRequiredWorkerBlock(unmet []subagents.AgentSnapshot) string {
215 var sb strings.Builder
216 sb.WriteString(requiredWorkerContextHeader)
217 sb.WriteString("\nTerminal completion is blocked by required delegated work:")
218 for _, snapshot := range unmet {
219 fmt.Fprintf(&sb, "\n- %s: %s — %s", snapshot.ID, snapshot.Requirement.State, strings.TrimSpace(snapshot.Task))
220 if snapshot.Requirement.ErrorCode != "" {
221 fmt.Fprintf(&sb, " (%s; diagnostic: %s)", snapshot.Requirement.ErrorCode, subagents.ResultRef(snapshot.ID))
222 }
223 }
224 if hasIndeterminateRequirement(unmet) {
225 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.")
226 } else {
227 sb.WriteString("\nRetry unresolved work with subagent_resume. Do not claim completion. The user may explicitly waive a terminal worker with /subagents rm.")
228 }
229 return sb.String()
230}
231
232func hasIndeterminateRequirement(required []subagents.AgentSnapshot) bool {
233 for _, snapshot := range required {
234 if snapshot.Requirement.State == subagents.RequirementIndeterminate {
235 return true
236 }
237 }
238 return false
239}
240
241func hasUnmetRequirement(required []subagents.AgentSnapshot) bool {
242 for _, snapshot := range required {
243 if snapshot.Requirement.Unmet() {
244 return true
245 }
246 }
247 return false
248}
249
250func joinRequiredContext(first, second string) string {
251 first = strings.TrimSpace(first)
252 second = strings.TrimSpace(second)
253 switch {
254 case first == "":
255 return second
256 case second == "":
257 return first
258 default:
259 return first + "\n\n" + second
260 }
261}
262
263func indentRequiredOutput(text, prefix string, maxRunes int) string {
264 text = strings.TrimSpace(text)
265 runes := []rune(text)
266 if maxRunes > 0 && len(runes) > maxRunes {
267 text = string(runes[:maxRunes]) + "\n…(truncated)"
268 }
269 return prefix + strings.ReplaceAll(text, "\n", "\n"+prefix)
270}