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