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