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