1package agent
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "strings"
8 "sync"
9
10 "github.com/bnema/zut/packages/agent/extensions"
11 "github.com/bnema/zut/packages/agent/subagents"
12 "github.com/bnema/zut/packages/agent/tools"
13 "github.com/bnema/zut/packages/core"
14 "github.com/bnema/zut/packages/provider"
15)
16
17// residentAgentRuntime runs a child's turns on one core.Agent and lets the
18// manager steer the running turn through the agent's queued-message boundary.
19type residentAgentRuntime struct {
20 agent *core.Agent
21 run func(context.Context, string) error
22
23 mu sync.Mutex
24 running bool
25}
26
27func (r *residentAgentRuntime) Run(ctx context.Context, prompt string) error {
28 r.mu.Lock()
29 r.running = true
30 r.mu.Unlock()
31 err := r.run(ctx, prompt)
32 for {
33 // A follow-up steered after the agent's last queue check would
34 // otherwise be silently left behind by a successful turn. Closing
35 // the steer window under the lock makes delivery all-or-nothing.
36 // On error or cancellation the queue is left for the child to
37 // drain and report as undelivered. The loop ends because each
38 // continuation drains the queue at its first step, and Steer only
39 // refills it on an explicit manager call.
40 r.mu.Lock()
41 if err != nil || ctx.Err() != nil || r.agent.QueuedMessageCount() == 0 {
42 r.running = false
43 r.mu.Unlock()
44 return err
45 }
46 r.mu.Unlock()
47 err = r.run(ctx, "")
48 }
49}
50
51func (r *residentAgentRuntime) Steer(prompt string) bool {
52 r.mu.Lock()
53 defer r.mu.Unlock()
54 if !r.running {
55 return false
56 }
57 return r.agent.QueueMessage(prompt, nil)
58}
59
60func (r *residentAgentRuntime) PendingSteers() int { return r.agent.QueuedMessageCount() }
61
62func (r *residentAgentRuntime) DrainSteers() []string {
63 drained := r.agent.DrainQueuedMessages()
64 out := make([]string, 0, len(drained))
65 for _, message := range drained {
66 out = append(out, message.Text)
67 }
68 return out
69}
70
71// newResidentChildRunner is the host-owned construction boundary for one
72// resident child. It resolves a fresh provider client and a fresh core.Agent
73// exactly once for the child's durable session; no subprocess configuration or
74// credential transfer is involved.
75func newResidentChildRunner(args Args, spec subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
76 if strings.TrimSpace(spec.SessionID) == "" {
77 return nil, fmt.Errorf("resident child %q has no session identity", spec.ID)
78 }
79 resolved, err := Resolve(args, true)
80 if err != nil {
81 return nil, fmt.Errorf("resolve resident child %q: %w", spec.ID, err)
82 }
83
84 registry, err := residentChildRegistry(resolved.ToolRegistry, spec.Tools)
85 if err != nil {
86 return nil, err
87 }
88 system := resolved.SystemPrompt
89 if extra := strings.TrimSpace(spec.SystemPrompt); extra != "" {
90 if spec.SystemPromptMode == "replace" {
91 system = extra
92 } else {
93 system = strings.TrimSpace(system) + "\n\n" + extra
94 }
95 }
96 agent := core.NewAgent(resolved.NewClient(), resolved.Model, system, registry)
97 // A resident child inherits the parent's resolved step limit, including the
98 // unlimited default. Its ceiling is the provider context window, with one
99 // compaction recovery per accepted turn, plus any explicit parent
100 // --max-steps. There is no child-only step budget. The explicit
101 // repetition-guard opt-out applies to child turns as well.
102 agent.MaxSteps = resolved.MaxSteps
103 agent.DisableRepetitionGuard = resolved.DisableRepetitionGuard
104 agent.ContextWindow = resolved.ContextWindow
105 agent.MaxTokens = resolved.MaxOutput
106 agent.Reasoning = resolved.Reasoning
107 agent.Temperature = resolved.Temperature
108 agent.FastMode = resolved.FastMode
109 var baseline subagents.ResidentUsageSnapshot
110 if journal != nil {
111 baseline = journal.ConfigureUsage(resolved.ContextWindow, resolved.AuthMethod == "oauth")
112 agent.SeedCost(baseline.Usage)
113 }
114 // The nudge reads the agent's own per-turn usage gauge. It stays advisory:
115 // the provider context window is the hard ceiling on a child turn.
116 configureResidentContextNudge(agent, resolved.ContextWindow)
117 rootCacheID := strings.TrimSpace(spec.RootCacheID)
118 if rootCacheID == "" {
119 // Journals written before root cache identity was persisted can still
120 // resume safely: their immediate parent was the only available root.
121 rootCacheID = strings.TrimSpace(spec.ParentSessionID)
122 if rootCacheID == "" {
123 rootCacheID = spec.SessionID
124 }
125 }
126 if err := agent.BindRequestIdentity(rootCacheID, spec.SessionID); err != nil {
127 return nil, fmt.Errorf("bind resident child %q request identity: %w", spec.ID, err)
128 }
129 if journal != nil {
130 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
131 if err != nil {
132 return nil, fmt.Errorf("restore resident child %q transcript: %w", spec.ID, err)
133 }
134 if len(messages) > 0 {
135 agent.SetMessages(messages)
136 }
137 }
138 limit := agent.MaxSteps
139 // run executes one accepted prompt, or continues the transcript to deliver
140 // late steered follow-ups when prompt is empty.
141 run := func(ctx context.Context, prompt string) error {
142 turnCtx, cancel := context.WithCancel(ctx)
143 defer cancel()
144 var journalErr error
145 var journalMu sync.Mutex
146 checkJournal := func() error {
147 journalMu.Lock()
148 defer journalMu.Unlock()
149 if journalErr == nil {
150 return nil
151 }
152 return fmt.Errorf("persist resident child transcript: %w", journalErr)
153 }
154 // usedSteps is the step high-water mark of this accepted turn. The sink
155 // runs on the runner goroutine, so it needs no lock; the mark bounds the
156 // continuation after a compaction recovery so an explicit parent limit
157 // is never replenished.
158 usedSteps := 0
159 sink := func(event core.AgentEvent) {
160 if turnStart, ok := event.(core.EvTurnStart); ok && turnStart.Step > usedSteps {
161 usedSteps = turnStart.Step
162 }
163 if err := journal.RecordAgentEvent(event); err != nil {
164 journalMu.Lock()
165 if journalErr == nil {
166 journalErr = err
167 cancel()
168 }
169 journalMu.Unlock()
170 }
171 }
172 var err error
173 if prompt == "" {
174 err = agent.Continue(turnCtx, sink)
175 } else {
176 err = agent.Prompt(turnCtx, prompt, nil, sink)
177 }
178 if journalFailure := checkJournal(); journalFailure != nil {
179 return journalFailure
180 }
181 if err != nil && turnCtx.Err() == nil && provider.IsContextOverflowError(err) {
182 err = recoverResidentContextOverflow(turnCtx, agent, journal, sink, limit, usedSteps, err)
183 if journalFailure := checkJournal(); journalFailure != nil {
184 return journalFailure
185 }
186 }
187 return err
188 }
189 return &residentAgentRuntime{agent: agent, run: run}, nil
190}
191
192// A compaction recovery never rewrites a transcript with nothing worth
193// summarizing: the child task and the current exchange must survive verbatim.
194const (
195 residentContextRecoveryMinimumMessages = 4
196 residentContextRecoveryKeepTail = 2
197)
198
199// recoverResidentContextOverflow gives one accepted resident turn a second
200// chance after the provider rejected it for exceeding the model context
201// window. It compacts the transcript, journals the replacement transcript, and
202// continues the same turn. Recovery happens at most once per call, and any
203// compaction or persistence failure restores the exact pre-compaction
204// transcript before returning the original error.
205func recoverResidentContextOverflow(ctx context.Context, agent *core.Agent, journal *subagents.ResidentJournal, sink func(core.AgentEvent), limit, usedSteps int, overflowErr error) error {
206 if agent == nil || journal == nil {
207 return overflowErr
208 }
209 before := agent.Messages()
210 if len(before) < residentContextRecoveryMinimumMessages {
211 // Summarizing a transcript this short would replace the instruction
212 // rather than shrink prior work.
213 return overflowErr
214 }
215 if _, err := agent.CompactWithEvents(ctx, residentContextRecoveryKeepTail, sink); err != nil {
216 agent.RestoreMessages(before)
217 return errors.Join(overflowErr, fmt.Errorf("compact resident child transcript: %w", err))
218 }
219 if err := journal.RecordCompacted(residentCheckpointMessages(agent.Messages())); err != nil {
220 // Memory must not run ahead of durable history: a later resume would
221 // otherwise replay the pre-compaction transcript.
222 agent.RestoreMessages(before)
223 return errors.Join(overflowErr, fmt.Errorf("persist resident child compaction: %w", err))
224 }
225 // The summarization request's usage describes the pre-compaction
226 // transcript. It must not arm the next reminder, which would report a
227 // false percentage for the compacted transcript the continuation sends.
228 agent.SeedLastTurnUsage(provider.Usage{})
229 if limit > 0 {
230 remaining := limit - usedSteps
231 if remaining <= 0 {
232 // An explicit parent limit is already spent. The checkpoint stays
233 // durable so an explicit resume starts from the compacted
234 // transcript, but this turn is not continued.
235 return overflowErr
236 }
237 agent.MaxSteps = remaining
238 defer func() { agent.MaxSteps = limit }()
239 }
240 return agent.Continue(ctx, sink)
241}
242
243// residentCheckpointMessages builds the durable replacement transcript of a
244// compaction. Host-authored internal context is dropped: it belongs to the
245// live request, and a replayed reminder would survive as a stale instruction
246// that the resumed child would still obey.
247func residentCheckpointMessages(messages []provider.Message) []provider.Message {
248 checkpoint := make([]provider.Message, 0, len(messages))
249 for _, message := range messages {
250 if core.IsInternalContextMessage(message) {
251 continue
252 }
253 checkpoint = append(checkpoint, message)
254 }
255 return checkpoint
256}
257
258// residentContextReminderBands are the context-usage percentages that arm the
259// resident reminder. Each band fires at most once per child runner, so a child
260// whose prompt keeps growing toward the window is told to wrap up again
261// without turning the reminder into a budget.
262var residentContextReminderBands = [...]int{85, 90, 95}
263
264func configureResidentContextNudge(agent *core.Agent, contextMax int) {
265 if agent == nil {
266 return
267 }
268 // deliveredBand is owned by the runner closure and never reset from
269 // Resume or journal paths. The injected developer-context message persists
270 // via appendDynamicContext and is not replayed verbatim from the child
271 // journal after restart; restarting a child rests the ladder.
272 deliveredBand := -1
273 agent.BeforeTurnContext = func(_ context.Context, _ int) (bool, string, string) {
274 if contextMax <= 0 {
275 return true, "", ""
276 }
277 usage := agent.LastTurnUsage()
278 used := usage.InputTokens + usage.CacheReadTokens + usage.CacheWriteTokens
279 band := residentContextReminderBand(used, contextMax)
280 if band <= deliveredBand {
281 return true, "", ""
282 }
283 deliveredBand = band
284 return true, "", fmt.Sprintf("Context usage is at %d%% of the model window. Finish the current task, report results, limits, and remaining verifications, and do not start broad new work.", used*100/contextMax)
285 }
286}
287
288// residentContextReminderBand returns the highest armed band crossed by the
289// current prompt size, or -1 while usage stays below the first band. A direct
290// jump to a higher band delivers one reminder for that band.
291func residentContextReminderBand(used, contextMax int) int {
292 if contextMax <= 0 {
293 return -1
294 }
295 band := -1
296 for _, candidate := range residentContextReminderBands {
297 if used*100 >= contextMax*candidate {
298 band = candidate
299 }
300 }
301 return band
302}
303
304// residentChildArgs applies the complete durable child specification without
305// rediscovering a mutable profile file. That preserves the profile inheritance
306// decision accepted at spawn time across explicit resume and restart.
307func residentChildArgs(args Args, parentProvider string, spec subagents.ResidentChildSpec) Args {
308 next := args
309 // Primary delegation policy belongs to the parent, not its workers. A child
310 // receives its accepted profile instructions and exact child tool list.
311 next.Orchestrate = false
312 next.ResidentChild = true
313 // An explicit CLI key belongs to the provider that resolved the parent.
314 // A child can select another provider, but it must resolve that provider's
315 // own credential rather than forwarding or persisting the parent's key.
316 if canonicalProvider(parentProvider) != canonicalProvider(spec.Provider) {
317 next.APIKey = ""
318 }
319 next.Provider = spec.Provider
320 next.BaseURL = spec.BaseURL
321 next.InsecureTLS = spec.InsecureTLS
322 next.Model = spec.Model
323 next.Reasoning = spec.Reasoning
324 next.FastMode = spec.FastMode
325 next.FastModeSet = true
326 if strings.TrimSpace(spec.Workspace) != "" {
327 next.CWD = spec.Workspace
328 }
329 if spec.InheritSkills != nil && !*spec.InheritSkills {
330 next.NoSkill = true
331 }
332 if spec.InheritProjectContext != nil && !*spec.InheritProjectContext {
333 next.NoContextFiles = true
334 }
335 return next
336}
337
338// residentChildRegistry applies an already validated, exact child tool list.
339// It always strips delegation and host-goal mutation tools: child agents are
340// independent workers, not nested orchestrators or goal owners.
341func residentChildRegistry(catalogue core.Registry, names []string) (core.Registry, error) {
342 registry := make(core.Registry, len(names))
343 for _, name := range names {
344 name = strings.TrimSpace(name)
345 if name == "" {
346 return nil, fmt.Errorf("resident child has an empty tool name")
347 }
348 if name == tools.SubagentToolName || name == "update_goal" || name == tools.PlanToolName {
349 return nil, fmt.Errorf("resident child tool %q is not allowed", name)
350 }
351 tool, ok := catalogue[name]
352 if !ok {
353 return nil, fmt.Errorf("resident child declares unavailable tool %q", name)
354 }
355 if extensions.IsInteractiveTool(tool) {
356 return nil, fmt.Errorf("resident child tool %q is interactive and not allowed", name)
357 }
358 registry[name] = tool
359 }
360 return registry, nil
361}