1package core
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "strings"
10 "sync"
11 "time"
12 "unicode/utf8"
13
14 "github.com/bnema/zut/packages/provider"
15 "github.com/google/uuid"
16)
17
18var (
19 ErrStreamIdleTimeout = errors.New("provider stream idle timeout")
20 ErrMaxSteps = errors.New("agent model step limit exceeded")
21)
22
23const (
24 DefaultStreamIdleTimeout = 5 * time.Minute
25 DefaultStreamMaxRetries = 5
26)
27
28// QueuedMessage is a user prompt waiting for the next safe model-call boundary.
29type QueuedMessage struct {
30 Text string
31 Images []provider.ImageBlock
32 // HostEvent distinguishes host evidence from ordinary input so hosts can
33 // retain it after a failed turn. Explicit cancellation may still discard it.
34 HostEvent bool
35}
36
37type queuedMessage struct {
38 message QueuedMessage
39 accepted time.Time
40}
41
42// requestOpenError marks a failure returned before a provider stream exists.
43// Provider clients own retries for this phase; the agent retry budget is only
44// for reconnecting a stream that opened and then failed.
45type requestOpenError struct{ err error }
46
47func (e *requestOpenError) Error() string { return e.err.Error() }
48func (e *requestOpenError) Unwrap() error { return e.err }
49
50// Agent is a stateful conversation bound to a provider client, a model,
51// and a set of tools.
52type Agent struct {
53 Client provider.Client
54 Model string
55 // System is mutex-protected; use SetSystemPrompt to write it and
56 // PromptConfig to read it.
57 System string
58 // Tools is mutex-protected; use SetTools to write it and PromptConfig
59 // or ToolsSnapshot to read it.
60 Tools Registry
61 MaxSteps int
62 // DisableRepetitionGuard opts out of repeated-content detection for
63 // intentional polling workflows.
64 DisableRepetitionGuard bool
65 Reasoning string
66 Temperature *float32
67 // FastMode requests the OpenAI fast service tier. The provider boundary
68 // rejects it for providers that do not support that contract.
69 FastMode bool
70
71 // ContextWindow is the effective input context capacity retained from
72 // model resolution. Hosts use it for proactive compaction even when the
73 // model was synthesized for an open-catalog provider.
74 ContextWindow int
75
76 // MaxTokens caps the model's output tokens per turn. Zero leaves
77 // the field unset on the provider request, letting each provider
78 // apply its own default (which can be conservative, e.g. Bedrock
79 // defaults to 4096, truncating long writes/edits). Hosts populate
80 // this from the resolved model's MaxOutput so large single-turn
81 // responses aren't silently cut off with stopReason=length.
82 MaxTokens int
83
84 // BeforeToolExecute, if set, is called immediately before each
85 // tool runs. Returning (allowed=false, reason) short-circuits
86 // the call with an error result containing reason. Optionally,
87 // returning a non-nil modifiedArgs replaces the JSON args the
88 // tool will see, which lets guards redact / augment / patch the
89 // model's request without rewriting the transcript. Empty or
90 // malformed modifiedArgs is ignored.
91 BeforeToolExecute func(call provider.ToolCallBlock) (allowed bool, reason string, modifiedArgs json.RawMessage)
92
93 // BeforeTurn, if set, is called before each turn's model call.
94 // Returning (allowed=false, reason) aborts the turn; reason is
95 // surfaced as an assistant-like status line. Used for rate-
96 // limiting, business-hour gates, and deny-by-default setups.
97 BeforeTurn func(step int) (allowed bool, reason string)
98
99 // BeforeTurnContext is the richer turn-start hook. It receives the
100 // current turn context so blocking interception can honor cancellation.
101 // It may return a bounded, non-transcript context string that is appended
102 // to the provider request's system prompt for this turn only. When set, it
103 // replaces BeforeTurn so a host can combine approval and context
104 // collection into one lifecycle round trip.
105 BeforeTurnContext func(ctx context.Context, step int) (allowed bool, reason, context string)
106
107 // HostContext returns bounded host-owned context for the current turn. It is
108 // composed with extension context instead of replacing extension hooks.
109 HostContext func(ctx context.Context) string
110
111 // CommitToolResult, if set, commits extension/tool metadata immediately
112 // after each tool call completes. Returning an error converts the result to
113 // an error before it is appended to the transcript or sent to the provider.
114 CommitToolResult func(id string, result ToolResult) error
115 // OnToolResult observes the final result after CommitToolResult succeeds or
116 // converts it to an error.
117 OnToolResult func(id string, result ToolResult)
118
119 // BeforeAssistantMessage, if set, is called after the model's
120 // final assistant message is assembled but before it's appended
121 // to the transcript. Returning (allowed=false) suppresses both
122 // the transcript append and the UI event. A non-empty
123 // replacement rewrites the visible text for the user while
124 // leaving the model's original text in the transcript (so the
125 // model can still see what it said in subsequent turns).
126 BeforeAssistantMessage func(text string) (allowed bool, reason, replacement string)
127
128 // MaxRetries controls reconnection attempts for transient failures after the
129 // provider stream opens. Zero disables this retry layer.
130 // RetryBaseDelay is doubled for each attempt; zero uses 2s.
131 MaxRetries int
132 RetryBaseDelay time.Duration
133 // StreamIdleTimeout cancels an individual provider stream when it emits no
134 // events for this duration. Zero leaves the stream unbounded.
135 StreamIdleTimeout time.Duration
136 streamIdleTimer func(time.Duration) (<-chan time.Time, func(), func())
137
138 // OnEvent, if set, mirrors every AgentEvent the loop emits to
139 // this callback in addition to the per-Prompt sink. Used by the
140 // extension manager to fan events out to subscribed extensions
141 // without each caller having to compose sinks manually.
142 OnEvent func(AgentEvent)
143
144 // OnMessageAppended, if set, fires every time a message is
145 // appended to the in-memory transcript by the agent loop — the
146 // initial user prompt, each finalised assistant message, and
147 // each tool-results message (plus the synthetic OpenAI image
148 // mirror, if any). Hosts wire this to the on-disk session so
149 // that turns are durable as soon as they happen, instead of
150 // only being flushed on a clean exit.
151 OnMessageAppended func(provider.Message)
152
153 // OnUsage, if set, fires after every turn's usage row arrives,
154 // carrying the cumulative usage for the session. Hosts wire
155 // this to the on-disk session so the persisted total stays
156 // current and a crash recovers the right cost figure.
157 OnUsage func(cumulative provider.Usage)
158
159 // OnRetryLifecycle receives sanitized retry audit records. It never
160 // receives provider error text or response bodies.
161 OnRetryLifecycle func(RetryLifecycleRecord)
162
163 // OnTranscriptCompacted, if set, fires after Compact replaces the
164 // in-memory transcript with the synthetic summary plus kept tail.
165 // Hosts wire this to append an explicit compaction checkpoint to
166 // the session log; per-message append hooks do not fire for this
167 // wholesale transcript replacement.
168 OnTranscriptCompacted func(messages []provider.Message)
169
170 mu sync.Mutex
171 messages []provider.Message
172 // plan is the agent-owned task checklist. It is mutex-protected; use
173 // SetPlan to seed it and CurrentPlan to read it.
174 plan []PlanStep
175 // rev increments whenever the transcript slice is replaced or a
176 // message is appended. The TUI uses it as a cheap redraw cache key
177 // so editor-only typing doesn't copy/rebuild a long transcript on
178 // every keypress.
179 rev uint64
180 cost CostTracker
181
182 // queued holds user messages submitted while the agent is busy.
183 // The loop appends them as normal user messages at safe
184 // boundaries: before the next model call after a tool batch, or
185 // after a text-only assistant turn finishes. It never interrupts
186 // a running tool or cancels an in-flight provider request.
187 queued []queuedMessage
188 // deniedToolCall tracks whether any tool invocation was refused at the
189 // execution boundary (guard or confirmation denial) during the current
190 // top-level runLoop invocation. A denied invocation must not gain a
191 // recovery prompt urging more actions.
192 deniedToolCall bool
193 // repetition state is agent-owned rather than transcript-derived so it
194 // survives compaction and automated Prompt/Continue boundaries.
195 repetition repetitionGuardState
196 timeContext agentTimeContext
197 hasSessionTimeContext bool
198 // sessionID identifies this conversation thread. cacheSessionID identifies
199 // the root prompt-cache affinity and may be shared by resident child
200 // threads. Hosts bind both before the first provider request; unbound
201 // agents allocate one shared ephemeral value for the process lifetime.
202 sessionID string
203 cacheSessionID string
204 providerRequestStarted bool
205}
206
207// NewAgent returns an Agent with sensible defaults.
208func NewAgent(client provider.Client, model, system string, tools Registry) *Agent {
209 return &Agent{
210 Client: client,
211 Model: model,
212 System: system,
213 Tools: tools,
214 MaxSteps: 0, // 0 = unlimited
215 MaxRetries: DefaultStreamMaxRetries,
216 RetryBaseDelay: 2 * time.Second,
217 StreamIdleTimeout: DefaultStreamIdleTimeout,
218 timeContext: newAgentTimeContext(time.Now()),
219 }
220}
221
222// BindSessionID binds a root conversation's durable session as both its
223// thread identity and prompt-cache affinity. Resident children use
224// BindRequestIdentity to retain their own thread while sharing their root's
225// cache affinity.
226func (a *Agent) BindSessionID(id string) error {
227 return a.BindRequestIdentity(id, id)
228}
229
230// BindRequestIdentity assigns the stable root cache and thread identities
231// before the first provider request.
232func (a *Agent) BindRequestIdentity(cacheSessionID, threadID string) error {
233 cacheSessionID = strings.TrimSpace(cacheSessionID)
234 threadID = strings.TrimSpace(threadID)
235 if cacheSessionID == "" || threadID == "" {
236 return errors.New("cache session ID and thread ID are required")
237 }
238 a.mu.Lock()
239 defer a.mu.Unlock()
240 if a.providerRequestStarted {
241 return errors.New("cannot change request identity after the first provider request")
242 }
243 a.cacheSessionID = cacheSessionID
244 a.sessionID = threadID
245 return nil
246}
247
248// SessionID returns the stable conversation-thread identity, allocating an
249// ephemeral identity when persistence is disabled and the host did not bind
250// one during runtime assembly.
251func (a *Agent) SessionID() string {
252 a.mu.Lock()
253 defer a.mu.Unlock()
254 a.ensureRequestIdentityLocked()
255 return a.sessionID
256}
257
258// CacheSessionID returns the stable root cache-affinity identity.
259func (a *Agent) CacheSessionID() string {
260 a.mu.Lock()
261 defer a.mu.Unlock()
262 a.ensureRequestIdentityLocked()
263 return a.cacheSessionID
264}
265
266func (a *Agent) ensureRequestIdentityLocked() {
267 if a.sessionID == "" {
268 a.sessionID = uuid.NewString()
269 }
270 if a.cacheSessionID == "" {
271 a.cacheSessionID = a.sessionID
272 }
273}
274
275func (a *Agent) beginTurn() provider.RequestContext {
276 a.mu.Lock()
277 a.ensureRequestIdentityLocked()
278 context := provider.RequestContext{CacheSessionID: a.cacheSessionID, ThreadID: a.sessionID, TurnID: uuid.NewString()}
279 a.mu.Unlock()
280 return context
281}
282
283func (a *Agent) beginProviderRequest() {
284 a.mu.Lock()
285 a.providerRequestStarted = true
286 a.mu.Unlock()
287}
288
289// SetFastMode changes the service-tier preference for the next model call.
290// It is safe to use while a turn is in flight; the active turn keeps the
291// value it already captured.
292func (a *Agent) SetFastMode(enabled bool) {
293 a.mu.Lock()
294 a.FastMode = enabled
295 a.mu.Unlock()
296}
297
298// FastModeEnabled returns the service-tier preference for a model call.
299func (a *Agent) FastModeEnabled() bool {
300 a.mu.Lock()
301 defer a.mu.Unlock()
302 return a.FastMode
303}
304
305// QueueMessage queues a text and image prompt to be injected as a user message
306// at the next safe boundary of the active agent loop. It is non-blocking in the
307// sense that it never waits for model/tool work; it only takes the transcript
308// mutex briefly. Prompts without text or images are ignored.
309func (a *Agent) QueueMessage(text string, images []provider.ImageBlock) bool {
310 return a.QueuePrompt(QueuedMessage{Text: text, Images: images})
311}
312
313// QueuePrompt queues a prompt with its host provenance at the next safe boundary.
314func (a *Agent) QueuePrompt(prompt QueuedMessage) bool {
315 prompt.Text = strings.TrimSpace(prompt.Text)
316 if prompt.Text == "" && len(prompt.Images) == 0 {
317 return false
318 }
319 message := cloneQueuedMessage(prompt)
320 a.mu.Lock()
321 a.queued = append(a.queued, queuedMessage{message: message, accepted: time.Now()})
322 a.mu.Unlock()
323 return true
324}
325
326// PendingQueuedMessages returns a snapshot of user messages waiting
327// to be injected. Used by hosts to render the visible "sliding in"
328// chips without consuming them.
329func (a *Agent) PendingQueuedMessages() []QueuedMessage {
330 a.mu.Lock()
331 defer a.mu.Unlock()
332 out := make([]QueuedMessage, len(a.queued))
333 for i, queued := range a.queued {
334 out[i] = cloneQueuedMessage(queued.message)
335 }
336 return out
337}
338
339// QueuedMessageCount returns the number of messages waiting to be
340// injected at the next safe boundary.
341func (a *Agent) QueuedMessageCount() int {
342 a.mu.Lock()
343 defer a.mu.Unlock()
344 return len(a.queued)
345}
346
347// PopQueuedMessage removes and returns the most recently queued
348// message. Hosts use this for the slide-back keybinding.
349func (a *Agent) PopQueuedMessage() (QueuedMessage, bool) {
350 a.mu.Lock()
351 defer a.mu.Unlock()
352 n := len(a.queued)
353 if n == 0 {
354 return QueuedMessage{}, false
355 }
356 message := cloneQueuedMessage(a.queued[n-1].message)
357 a.queued = a.queued[:n-1]
358 return message, true
359}
360
361// DrainQueuedMessages discards and returns every queued message.
362// Hosts use this on explicit cancel/clear so stale follow-ups do
363// not run after the user aborted the turn.
364func (a *Agent) DrainQueuedMessages() []QueuedMessage {
365 queued := a.drainQueuedMessages()
366 out := make([]QueuedMessage, len(queued))
367 for i, message := range queued {
368 out[i] = cloneQueuedMessage(message.message)
369 }
370 return out
371}
372
373func cloneQueuedMessage(message QueuedMessage) QueuedMessage {
374 clone := QueuedMessage{Text: message.Text, HostEvent: message.HostEvent}
375 if len(message.Images) == 0 {
376 return clone
377 }
378 clone.Images = make([]provider.ImageBlock, len(message.Images))
379 for i, image := range message.Images {
380 clone.Images[i] = image
381 clone.Images[i].Data = append([]byte(nil), image.Data...)
382 }
383 return clone
384}
385
386func (a *Agent) drainQueuedMessages() []queuedMessage {
387 a.mu.Lock()
388 defer a.mu.Unlock()
389 out := append([]queuedMessage(nil), a.queued...)
390 a.queued = nil
391 return out
392}
393
394func (a *Agent) appendQueuedAsUser(messages []queuedMessage, sink func(AgentEvent)) {
395 for _, queued := range messages {
396 accepted := queued.accepted
397 if accepted.IsZero() {
398 accepted = time.Now()
399 }
400 content := make([]provider.Content, 0, 1+len(queued.message.Images))
401 if queued.message.Text != "" {
402 content = append(content, provider.TextBlock{Text: queued.message.Text})
403 }
404 for _, image := range queued.message.Images {
405 content = append(content, image)
406 }
407 msg := provider.Message{
408 Role: provider.RoleUser,
409 Content: content,
410 Time: accepted,
411 }
412 a.mu.Lock()
413 if !queued.message.HostEvent {
414 a.repetition = repetitionGuardState{}
415 }
416 a.messages = append(a.messages, msg)
417 a.rev++
418 a.mu.Unlock()
419 a.fireMessageAppended(msg)
420 if sink != nil {
421 sink(EvUserMessage{Message: msg})
422 }
423 }
424}
425
426// Messages returns a copy of the current transcript.
427func (a *Agent) Messages() []provider.Message {
428 a.mu.Lock()
429 defer a.mu.Unlock()
430 out := make([]provider.Message, len(a.messages))
431 copy(out, a.messages)
432 return out
433}
434
435// MessageCount returns the current transcript length without copying it.
436func (a *Agent) MessageCount() int {
437 a.mu.Lock()
438 defer a.mu.Unlock()
439 return len(a.messages)
440}
441
442// MessagesFrom returns a copy of the transcript suffix beginning at from and
443// the total transcript length observed under the same lock.
444func (a *Agent) MessagesFrom(from int) ([]provider.Message, int) {
445 a.mu.Lock()
446 defer a.mu.Unlock()
447 if from < 0 {
448 from = 0
449 }
450 if from > len(a.messages) {
451 from = len(a.messages)
452 }
453 out := make([]provider.Message, len(a.messages)-from)
454 copy(out, a.messages[from:])
455 return out, len(a.messages)
456}
457
458// Revision returns a monotonically increasing transcript version.
459// It is cheap to query and changes whenever Messages() would return
460// different transcript content because of append/set operations.
461func (a *Agent) Revision() uint64 {
462 a.mu.Lock()
463 defer a.mu.Unlock()
464 return a.rev
465}
466
467// SetTools swaps the tool registry. Used by /reload-ext to hand
468// the agent a fresh registry after extension subprocesses have been
469// respawned (and their freshly-registered tools merged in).
470func (a *Agent) SetTools(reg Registry) {
471 a.mu.Lock()
472 a.Tools = reg
473 a.mu.Unlock()
474}
475
476// SetSystemPrompt replaces the system prompt used for subsequent turns.
477func (a *Agent) SetSystemPrompt(system string) {
478 a.mu.Lock()
479 a.System = system
480 a.mu.Unlock()
481}
482
483// SetPromptConfig atomically replaces the system prompt and tool registry.
484// It returns the previous registry so the caller can close resources that are
485// no longer owned by the agent.
486func (a *Agent) SetPromptConfig(system string, tools Registry) Registry {
487 a.mu.Lock()
488 defer a.mu.Unlock()
489 oldTools := a.Tools
490 a.System = system
491 a.Tools = tools
492 return oldTools
493}
494
495// PromptConfig returns a consistent snapshot of the system prompt and tools
496// used to construct a provider request.
497func (a *Agent) PromptConfig() (string, Registry) {
498 a.mu.Lock()
499 defer a.mu.Unlock()
500 tools := make(Registry, len(a.Tools))
501 for name, tool := range a.Tools {
502 tools[name] = tool
503 }
504 return a.System, tools
505}
506
507// ToolsSnapshot returns a copy of the current tool registry.
508func (a *Agent) ToolsSnapshot() Registry {
509 a.mu.Lock()
510 defer a.mu.Unlock()
511 tools := make(Registry, len(a.Tools))
512 for name, tool := range a.Tools {
513 tools[name] = tool
514 }
515 return tools
516}
517
518// SetMessages replaces the transcript without changing repetition history.
519// Call ResetRepetitionGuard when replacing it for a new conversation.
520func (a *Agent) SetMessages(msgs []provider.Message) {
521 a.mu.Lock()
522 defer a.mu.Unlock()
523 a.messages = append(a.messages[:0], msgs...)
524 a.rev++
525}
526
527// RestoreMessages replaces a transcript without resetting loop detection.
528// Use it when rolling back a failed compaction/transcript replacement.
529func (a *Agent) RestoreMessages(msgs []provider.Message) {
530 a.mu.Lock()
531 defer a.mu.Unlock()
532 a.messages = append(a.messages[:0], msgs...)
533 a.rev++
534}
535
536// ResetRepetitionGuard clears repeated-content history for an explicit
537// user action without replacing the transcript.
538func (a *Agent) ResetRepetitionGuard() {
539 a.mu.Lock()
540 a.repetition = repetitionGuardState{}
541 a.mu.Unlock()
542}
543
544// AppendUserContext adds a user-role message to the transcript without
545// starting a model turn. Hosts use it for context gathered outside the agent
546// loop, such as the output of an explicitly invoked shell command.
547func (a *Agent) AppendUserContext(text string, meta map[string]string) {
548 msg := provider.Message{
549 Role: provider.RoleUser,
550 Content: []provider.Content{provider.TextBlock{Text: text}},
551 Time: time.Now(),
552 Meta: meta,
553 }
554 a.mu.Lock()
555 a.messages = append(a.messages, msg)
556 a.rev++
557 a.mu.Unlock()
558 a.fireMessageAppended(msg)
559}
560
561// Cost returns the cumulative usage.
562func (a *Agent) Cost() provider.Usage {
563 a.mu.Lock()
564 defer a.mu.Unlock()
565 return a.cost.Total
566}
567
568func (a *Agent) addUsage(usage provider.Usage) provider.Usage {
569 a.mu.Lock()
570 defer a.mu.Unlock()
571 return a.cost.Add(usage)
572}
573
574// SeedCost sets the cumulative usage as a baseline before the first
575// turn runs. Used when transferring state from another agent (model
576// or provider switch) so the running cost meter doesn't reset to 0.
577func (a *Agent) SeedCost(u provider.Usage) {
578 a.mu.Lock()
579 defer a.mu.Unlock()
580 a.cost.Seed(u)
581}
582
583// LastTurnUsage returns the per-turn usage of the most recent
584// completed turn. Drives the "context used" gauge in the status bar
585// without waiting for the next turn to land.
586func (a *Agent) LastTurnUsage() provider.Usage {
587 a.mu.Lock()
588 defer a.mu.Unlock()
589 return a.cost.LastTurn
590}
591
592// SeedLastTurnUsage primes the per-turn snapshot. Used on resume so
593// the gauge reflects the prompt size of the last turn in the session
594// file instead of starting at zero.
595func (a *Agent) SeedLastTurnUsage(u provider.Usage) {
596 a.mu.Lock()
597 defer a.mu.Unlock()
598 a.cost.LastTurn = u
599}
600
601// fireMessageAppended invokes OnMessageAppended without holding the
602// agent mutex, so the host's persistence callback can take its own
603// locks without deadlocking the agent loop. Tolerates a nil hook so
604// non-persisting callers (tests, RPC mode) don't have to set it.
605func (a *Agent) fireMessageAppended(m provider.Message) {
606 if a.OnMessageAppended != nil {
607 a.OnMessageAppended(m)
608 }
609}
610
611// Prompt sends a new user message and runs the agent loop until the model
612// stops or an error occurs. It resets repeated-content tracking for the new
613// request. Events are delivered via sink in order. sink must not block the
614// caller for long; buffer as needed.
615func (a *Agent) Prompt(ctx context.Context, text string, images []provider.ImageBlock, sink func(AgentEvent)) error {
616 if text == "" && len(images) == 0 {
617 return errors.New("prompt is empty")
618 }
619 if sink == nil {
620 sink = func(AgentEvent) {}
621 }
622 sink = a.wrapSink(sink)
623 content := []provider.Content{}
624 if text != "" {
625 content = append(content, provider.TextBlock{Text: text})
626 }
627 for _, img := range images {
628 content = append(content, img)
629 }
630 user := provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
631
632 a.mu.Lock()
633 a.messages = append(a.messages, user)
634 a.rev++
635 a.repetition = repetitionGuardState{}
636 a.mu.Unlock()
637 a.fireMessageAppended(user)
638 sink(EvUserMessage{Message: user})
639
640 return a.runLoop(ctx, sink, a.beginTurn())
641}
642
643// Continue runs the agent loop against the existing transcript. Used
644// after appending tool results manually or to retry.
645func (a *Agent) Continue(ctx context.Context, sink func(AgentEvent)) error {
646 if sink == nil {
647 sink = func(AgentEvent) {}
648 }
649 sink = a.wrapSink(sink)
650 return a.runLoop(ctx, sink, a.beginTurn())
651}
652
653// wrapSink composes the per-call sink with a.OnEvent (if set) so the
654// extension manager (or any other observer) sees every AgentEvent
655// without having to thread itself through every Prompt callsite.
656func (a *Agent) wrapSink(sink func(AgentEvent)) func(AgentEvent) {
657 if a.OnEvent == nil {
658 return sink
659 }
660 obs := a.OnEvent
661 return func(ev AgentEvent) {
662 obs(ev)
663 sink(ev)
664 }
665}
666
667func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent), requestContext provider.RequestContext) error {
668 a.resetDeniedToolCalls()
669 recoveryUsed := false
670 for step := 1; a.MaxSteps <= 0 || step <= a.MaxSteps; step++ {
671 // Messages queued while the agent was busy are delivered
672 // before the next model call. This is the safe boundary:
673 // any previous tool batch has already completed and its
674 // results have been appended, but no new provider request has
675 // started yet.
676 if pending := a.drainQueuedMessages(); len(pending) > 0 {
677 a.appendQueuedAsUser(pending, sink)
678 }
679 if a.repetitionStopped() {
680 sink(EvDone{})
681 return a.repetitiveLoopError()
682 }
683
684 sink(EvTurnStart{Step: step, TurnID: requestContext.TurnID})
685 turnContext := ""
686 if a.BeforeTurnContext != nil {
687 var allowed bool
688 var reason string
689 allowed, reason, turnContext = a.BeforeTurnContext(ctx, step)
690 if !allowed {
691 if reason == "" {
692 reason = "turn blocked by extension guard"
693 }
694 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
695 sink(EvDone{})
696 return nil
697 }
698 } else if a.BeforeTurn != nil {
699 if allowed, reason := a.BeforeTurn(step); !allowed {
700 if reason == "" {
701 reason = "turn blocked by extension guard"
702 }
703 sink(EvTurnEnd{Stop: provider.StopError, Err: fmt.Errorf("%s", reason)})
704 sink(EvDone{})
705 return nil
706 }
707 }
708
709 hostContext := ""
710 if a.HostContext != nil {
711 hostContext = a.HostContext(ctx)
712 }
713 a.appendDynamicContext(hostContext, turnContext)
714
715 var (
716 stop provider.StopReason
717 assistantMsg provider.Message
718 err error
719 )
720 maxAttempts := max(a.MaxRetries+1, 1)
721 for attempt := 0; ; attempt++ {
722 stop, assistantMsg, err = a.oneTurn(ctx, sink, turnContext, requestContext, attempt, maxAttempts)
723 sink(EvTurnEnd{Stop: stop, Err: err})
724 if err == nil {
725 break
726 }
727 retryable := a.canRetryError(err, attempt)
728 var retryCtxErr error
729 if retryable {
730 retryCtxErr = ctx.Err()
731 if retryCtxErr != nil {
732 retryable = false
733 }
734 }
735 reason := classifyRetryReason(err)
736 if !errors.Is(err, context.Canceled) {
737 a.fireRetryLifecycle(RetryLifecycleRecord{
738 Event: RetryLifecycleRequestFailed,
739 Scope: RetryScopeAgent,
740 Attempt: attempt + 1,
741 MaxAttempts: maxAttempts,
742 Reason: reason,
743 Terminal: !retryable,
744 })
745 }
746 if retryCtxErr != nil {
747 return retryCtxErr
748 }
749 if !retryable {
750 break
751 }
752 a.dropLastAssistantMessage()
753 delay := a.retryDelay(attempt)
754 if retryErr := ctx.Err(); retryErr != nil {
755 return retryErr
756 }
757 a.fireRetryLifecycle(RetryLifecycleRecord{
758 Event: RetryLifecycleRetryScheduled,
759 Scope: RetryScopeAgent,
760 Attempt: attempt + 2,
761 MaxAttempts: maxAttempts,
762 Reason: reason,
763 DelayMS: delay.Milliseconds(),
764 })
765 sink(EvRetryScheduled{
766 Scope: RetryScopeAgent,
767 Attempt: attempt + 2,
768 MaxAttempts: maxAttempts,
769 Delay: delay,
770 })
771 if sleepErr := sleepRetry(ctx, delay); sleepErr != nil {
772 return sleepErr
773 }
774 }
775 if err != nil {
776 return err
777 }
778
779 if stop == provider.StopContinue {
780 if err := ctx.Err(); err != nil {
781 sink(EvDone{})
782 return err
783 }
784 if a.QueuedMessageCount() == 0 {
785 if decision := a.observeRepetition(assistantMessageFingerprint(assistantMsg), RepetitionKindAssistantMessage, ""); decision.stage != "" {
786 if a.handleRepetitionDecision(decision, sink) {
787 return a.repetitiveLoopError()
788 }
789 }
790 }
791 continue
792 }
793 if stop == provider.StopToolUse {
794 // Execute each tool call, append a single tool-results message, continue.
795 toolMsg, hadError := a.executeTools(ctx, assistantMsg, sink)
796 a.mu.Lock()
797 a.messages = append(a.messages, toolMsg)
798 a.rev++
799 // OpenAI's chat-completions tool message shape is text-centric.
800 // Vision models reliably consume images when they arrive as user
801 // content, so when a tool result contains images we mirror them
802 // into a synthetic user message immediately after the tool result.
803 // This keeps the transcript self-contained for providers that can
804 // see image blocks in tool messages while making OpenAI vision
805 // models actually receive the image bytes.
806 //
807 // The OpenAI Responses route ("openai-codex") has the same
808 // text-centric tool-output shape: a function_call_output only
809 // carries a string, so images in a tool result never reach the
810 // model. Both providers serialize images correctly when they
811 // arrive as user content, so the mirror covers them both.
812 var imageMirror provider.Message
813 if a.Client != nil && (a.Client.Name() == "openai" || a.Client.Name() == "openai-codex") {
814 if mirror := mirrorToolImagesAsUser(toolMsg); len(mirror.Content) > 0 {
815 a.messages = append(a.messages, mirror)
816 a.rev++
817 imageMirror = mirror
818 }
819 }
820 a.mu.Unlock()
821 a.fireMessageAppended(toolMsg)
822 if len(imageMirror.Content) > 0 {
823 a.fireMessageAppended(imageMirror)
824 }
825 // If context was cancelled during tool execution, bail out.
826 if err := ctx.Err(); err != nil {
827 sink(EvDone{})
828 return err
829 }
830 _ = hadError
831 for _, observation := range repetitionCallObservations(assistantMsg, toolMsg) {
832 decision := a.observeRepetition(observation.fingerprint, observation.kind, observation.toolName)
833 if decision.stage != "" && a.handleRepetitionDecision(decision, sink) {
834 return a.repetitiveLoopError()
835 }
836 }
837 continue
838 }
839
840 // If the assistant stopped without tool calls but a message was
841 // queued while it was speaking, loop once more so that message
842 // is appended and answered instead of waiting until a later
843 // top-level prompt.
844 if ctx.Err() == nil && a.QueuedMessageCount() > 0 {
845 continue
846 }
847 if stop == provider.StopEnd && !a.sawDeniedToolCall() {
848 if decision := a.observeRepetition(assistantMessageFingerprint(assistantMsg), RepetitionKindAssistantMessage, ""); decision.stage != "" {
849 if a.handleRepetitionDecision(decision, sink) {
850 return a.repetitiveLoopError()
851 }
852 continue
853 }
854 }
855
856 // Bounded normal-turn recovery: a successful normal stop whose
857 // current terminal assistant message carries no final answer gets
858 // one continuation per top-level invocation. Never on errors
859 // (err != nil returns above), cancellation, output/context
860 // limits, or a permission/confirmation refusal.
861 if stop == provider.StopEnd && !a.sawDeniedToolCall() {
862 if reason, incomplete := classifyTerminalAssistant(assistantMsg); incomplete {
863 if !recoveryUsed {
864 if err := ctx.Err(); err != nil {
865 sink(EvDone{})
866 return err
867 }
868 recoveryUsed = true
869 a.AppendUserContext(turnRecoveryPrompt, map[string]string{turnRecoveryMetaKey: "true"})
870 sink(EvTurnRecovery{Reason: reason, Attempt: 1})
871 continue
872 }
873 sink(EvDone{})
874 return ErrIncompleteTurn
875 }
876 }
877
878 // Terminal stop (end, length, error, aborted).
879 sink(EvDone{})
880 return nil
881 }
882 if a.MaxSteps > 0 {
883 sink(EvDone{})
884 return fmt.Errorf("%w: %d", ErrMaxSteps, a.MaxSteps)
885 }
886 return nil
887}
888
889func (a *Agent) canRetryError(err error, attempt int) bool {
890 if err == nil || a.MaxRetries <= 0 || attempt >= a.MaxRetries {
891 return false
892 }
893 var openErr *requestOpenError
894 if errors.As(err, &openErr) {
895 return false
896 }
897 // An idle timeout already bounds a full period without provider progress.
898 // Retrying it would restart that deadline for every attempt, allowing one
899 // silent model continuation to keep an agent running for more than thirty
900 // minutes under the default retry policy.
901 if errors.Is(err, ErrStreamIdleTimeout) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
902 return false
903 }
904 msg := strings.ToLower(err.Error())
905 if msg == "" || isNonRetryableProviderLimit(msg) {
906 return false
907 }
908 if isStreamNetworkError(err, msg) {
909 return true
910 }
911 needles := []string{
912 "overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
913 "429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
914 "service unavailable", "server error", "internal error", "ended without", "stream ended before",
915 "did not get a response", "timed out", "timeout", "terminated",
916 // OpenAI's ChatGPT/Codex backend returns this generic message (with a
917 // request ID) for transient server failures and explicitly says
918 // "You can retry your request".
919 "an error occurred while processing your request",
920 // Explicit retry guidance emitted by provider backends (OpenAI
921 // Responses, AWS Bedrock stream exceptions) with varying prefixes.
922 "you can retry your request", "try your request again", "please retry your request",
923 // Capacity messages from the ChatGPT/Codex backend, e.g.
924 // "Our servers are currently overloaded. Please try again later."
925 // The trailing advice also shows up on its own for transient
926 // capacity failures; usage/quota limits are filtered out above by
927 // isNonRetryableProviderLimit before this list is consulted.
928 "servers are currently overloaded", "servers are busy", "try again later",
929 }
930 for _, needle := range needles {
931 if strings.Contains(msg, needle) {
932 return true
933 }
934 }
935 return false
936}
937
938// isStreamNetworkError detects transport errors from an opened provider stream.
939// Keep typed EOF checks because WebSocket readers return io.EOF on a peer close,
940// while provider-specific transports may only preserve a textual error.
941func isStreamNetworkError(err error, msg string) bool {
942 if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
943 return true
944 }
945 for _, needle := range []string{
946 "connection", "network", "unexpected eof", "broken pipe", "socket hang up",
947 "transport failure", "upstream connect",
948 } {
949 if strings.Contains(msg, needle) {
950 return true
951 }
952 }
953 return false
954}
955
956func isNonRetryableProviderLimit(msg string) bool {
957 needles := []string{
958 "usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",
959 "available balance", "insufficient_quota", "out of budget", "quota exceeded", "billing",
960 }
961 for _, needle := range needles {
962 if strings.Contains(msg, needle) {
963 return true
964 }
965 }
966 return false
967}
968
969func classifyRetryReason(err error) RetryReason {
970 if err == nil {
971 return RetryReasonUnknown
972 }
973 msg := strings.ToLower(err.Error())
974 if isNonRetryableProviderLimit(msg) {
975 return RetryReasonQuota
976 }
977 if strings.Contains(msg, "overload") || strings.Contains(msg, "servers are busy") {
978 return RetryReasonOverload
979 }
980 if strings.Contains(msg, "rate limit") || strings.Contains(msg, "ratelimit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "429") {
981 return RetryReasonRateLimit
982 }
983 if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") {
984 return RetryReasonTimeout
985 }
986 if isStreamNetworkError(err, msg) {
987 return RetryReasonNetwork
988 }
989 for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} {
990 if strings.Contains(msg, needle) {
991 return RetryReasonContextWindow
992 }
993 }
994 for _, needle := range []string{"unauthorized", "authentication", "invalid api key", "invalid_api_key", "http 401", "http 403"} {
995 if strings.Contains(msg, needle) {
996 return RetryReasonAuth
997 }
998 }
999 for _, needle := range []string{"server error", "internal error", "service unavailable", "http 500", "http 502", "http 503", "http 504", "provider returned error"} {
1000 if strings.Contains(msg, needle) {
1001 return RetryReasonServer
1002 }
1003 }
1004 for _, needle := range []string{"bad request", "unsupported", "invalid parameter", "http 400", "http 404", "http 422"} {
1005 if strings.Contains(msg, needle) {
1006 return RetryReasonClient
1007 }
1008 }
1009 return RetryReasonUnknown
1010}
1011
1012func (a *Agent) fireRetryLifecycle(record RetryLifecycleRecord) {
1013 if a.OnRetryLifecycle != nil {
1014 a.OnRetryLifecycle(record)
1015 }
1016}
1017
1018func (a *Agent) retryDelay(attempt int) time.Duration {
1019 base := a.RetryBaseDelay
1020 if base <= 0 {
1021 base = 2 * time.Second
1022 }
1023 return base * time.Duration(1<<attempt)
1024}
1025
1026func sleepRetry(ctx context.Context, d time.Duration) error {
1027 if d <= 0 {
1028 return ctx.Err()
1029 }
1030 t := time.NewTimer(d)
1031 defer t.Stop()
1032 select {
1033 case <-ctx.Done():
1034 return ctx.Err()
1035 case <-t.C:
1036 return nil
1037 }
1038}
1039
1040func (a *Agent) dropLastAssistantMessage() {
1041 a.mu.Lock()
1042 defer a.mu.Unlock()
1043 if n := len(a.messages); n > 0 && a.messages[n-1].Role == provider.RoleAssistant {
1044 a.messages = a.messages[:n-1]
1045 a.rev++
1046 }
1047}
1048
1049// requestLifecycleSink adapts provider-owned pre-stream retries into core
1050// events without letting provider concerns leak into mode renderers.
1051type requestLifecycleSink struct {
1052 sink func(AgentEvent)
1053 observe func(RetryLifecycleRecord)
1054 provider string
1055 model string
1056 lastReason RetryReason
1057}
1058
1059func (s *requestLifecycleSink) CacheDiagnostics(diagnostics provider.CacheDiagnostics) {
1060 s.sink(EvCacheDiagnostics{
1061 Eligible: diagnostics.Eligible,
1062 Mode: diagnostics.Mode,
1063 Transport: diagnostics.Transport,
1064 Continuation: diagnostics.Continuation,
1065 })
1066}
1067
1068func (s *requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) {
1069 s.sink(EvRequestStarted{
1070 Provider: s.provider,
1071 Model: s.model,
1072 Scope: RetryScopeProvider,
1073 Attempt: attempt,
1074 MaxAttempts: maxAttempts,
1075 })
1076}
1077
1078func (s *requestLifecycleSink) RequestFailed(attempt, maxAttempts int, reason provider.RequestFailureReason, terminal bool) {
1079 s.lastReason = retryReasonFromProvider(reason)
1080 if s.observe != nil {
1081 s.observe(RetryLifecycleRecord{
1082 Event: RetryLifecycleRequestFailed,
1083 Scope: RetryScopeProvider,
1084 Attempt: attempt,
1085 MaxAttempts: maxAttempts,
1086 Reason: s.lastReason,
1087 Terminal: terminal,
1088 })
1089 }
1090}
1091
1092func (s *requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) {
1093 s.sink(EvRetryScheduled{
1094 Scope: RetryScopeProvider,
1095 Attempt: attempt,
1096 MaxAttempts: maxAttempts,
1097 Delay: delay,
1098 })
1099 if s.observe != nil {
1100 reason := s.lastReason
1101 if reason == "" {
1102 reason = RetryReasonUnknown
1103 }
1104 s.observe(RetryLifecycleRecord{
1105 Event: RetryLifecycleRetryScheduled,
1106 Scope: RetryScopeProvider,
1107 Attempt: attempt,
1108 MaxAttempts: maxAttempts,
1109 Reason: reason,
1110 DelayMS: delay.Milliseconds(),
1111 })
1112 }
1113 s.lastReason = ""
1114}
1115
1116func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason {
1117 switch reason {
1118 case provider.RequestFailureOverload:
1119 return RetryReasonOverload
1120 case provider.RequestFailureRateLimit:
1121 return RetryReasonRateLimit
1122 case provider.RequestFailureQuota:
1123 return RetryReasonQuota
1124 case provider.RequestFailureServer:
1125 return RetryReasonServer
1126 case provider.RequestFailureNetwork:
1127 return RetryReasonNetwork
1128 case provider.RequestFailureTimeout:
1129 return RetryReasonTimeout
1130 case provider.RequestFailureClient:
1131 return RetryReasonClient
1132 default:
1133 return RetryReasonUnknown
1134 }
1135}
1136
1137// oneTurn calls the LLM once, forwards events, returns the stop reason
1138// and the assembled assistant message (already appended to the transcript).
1139func newStreamIdleTimer(timeout time.Duration) (<-chan time.Time, func(), func()) {
1140 timer := time.NewTimer(timeout)
1141 stop := func() { timer.Stop() }
1142 reset := func() {
1143 if !timer.Stop() {
1144 select {
1145 case <-timer.C:
1146 default:
1147 }
1148 }
1149 timer.Reset(timeout)
1150 }
1151 return timer.C, stop, reset
1152}
1153
1154func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, requestContext provider.RequestContext, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
1155 fastMode := a.FastModeEnabled()
1156 if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {
1157 return provider.StopError, provider.Message{}, err
1158 }
1159 system, tools := a.PromptConfig()
1160 // Repair pairs before projecting the copied provider-input view. The
1161 // repair can add stub results for aborted calls; those results must also
1162 // remain in the request so every tool call still has a matching result.
1163 messages := repairToolUseResultPairs(a.Messages())
1164 requestLifecycle := &requestLifecycleSink{
1165 sink: sink,
1166 observe: a.fireRetryLifecycle,
1167 provider: a.Client.Name(),
1168 model: a.Model,
1169 }
1170 req := provider.Request{
1171 Model: a.Model,
1172 System: system,
1173 // Repair any dangling tool_use blocks before sending. A turn
1174 // aborted mid-flight (cancel, connection drop, ECONNREFUSED to a
1175 // dev server, etc.) can leave an assistant tool_use with no
1176 // matching tool_result in the live transcript. The load-time
1177 // repair in OpenSession only runs on restart, so without this the
1178 // next in-process request is rejected by providers like Anthropic
1179 // with "tool_use ids were found without tool_result blocks". The
1180 // repair is pure and a no-op on already-valid transcripts.
1181 Messages: projectProviderMessages(messages),
1182 Tools: tools.Specs(),
1183 Reasoning: a.Reasoning,
1184 FastMode: fastMode,
1185 MaxTokens: a.MaxTokens,
1186 Temperature: a.Temperature,
1187 Lifecycle: requestLifecycle,
1188 Context: requestContext,
1189 }
1190 sink(EvRequestStarted{
1191 Provider: a.Client.Name(),
1192 Model: a.Model,
1193 Scope: RetryScopeAgent,
1194 Attempt: attempt + 1,
1195 MaxAttempts: maxAttempts,
1196 })
1197 streamCtx, cancelStream := context.WithCancel(ctx)
1198 defer cancelStream()
1199 a.beginProviderRequest()
1200 stream, err := a.Client.Stream(streamCtx, req)
1201 if err != nil {
1202 return provider.StopError, provider.Message{}, &requestOpenError{err: err}
1203 }
1204 // A provider may already have queued events when cancellation wins. Keep a
1205 // consumer attached after this turn returns so bounded producer channels
1206 // cannot strand the provider goroutine and its response body.
1207 defer func() { go drainProviderStream(stream) }()
1208
1209 sink(EvAssistantStart{})
1210
1211 var (
1212 stop provider.StopReason
1213 finalErr error
1214 finalMsg provider.Message
1215 )
1216
1217 var idle <-chan time.Time
1218 var resetIdle func()
1219 if a.StreamIdleTimeout > 0 {
1220 newTimer := a.streamIdleTimer
1221 if newTimer == nil {
1222 newTimer = newStreamIdleTimer
1223 }
1224 var stopIdle func()
1225 idle, stopIdle, resetIdle = newTimer(a.StreamIdleTimeout)
1226 defer stopIdle()
1227 }
1228 handleStreamEvent := func(ev provider.Event) (done bool) {
1229 if resetIdle != nil {
1230 resetIdle()
1231 }
1232 switch e := ev.(type) {
1233 case provider.EventStart:
1234 // nothing
1235 case provider.EventTextDelta:
1236 sink(EvTextDelta{Delta: e.Delta})
1237 case provider.EventToolStart:
1238 sink(EvToolUseStart{ID: e.ID, Name: e.Name})
1239 case provider.EventToolArgs:
1240 sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
1241 case provider.EventToolEnd:
1242 sink(EvToolUseEnd{ID: e.ID})
1243 case provider.EventUsage:
1244 cum := a.addUsage(e.Usage)
1245 sink(EvUsage{Usage: e.Usage, Cumulative: cum})
1246 if a.OnUsage != nil {
1247 a.OnUsage(cum)
1248 }
1249 case provider.EventDone:
1250 stop = e.Stop
1251 finalErr = e.Err
1252 finalMsg = provider.WithMessageOrigin(e.Message, a.Client.Name(), req.Model)
1253 return true
1254 }
1255 return false
1256 }
1257 for {
1258 // Consume no more than one buffered event before honoring cancellation.
1259 select {
1260 case ev, ok := <-stream:
1261 if !ok || handleStreamEvent(ev) {
1262 goto streamDone
1263 }
1264 default:
1265 }
1266 if err := streamCtx.Err(); err != nil {
1267 return provider.StopError, finalMsg, err
1268 }
1269 select {
1270 case <-streamCtx.Done():
1271 return provider.StopError, finalMsg, streamCtx.Err()
1272 case <-idle:
1273 return provider.StopError, finalMsg, ErrStreamIdleTimeout
1274 case ev, ok := <-stream:
1275 if !ok || handleStreamEvent(ev) {
1276 goto streamDone
1277 }
1278 }
1279 }
1280streamDone:
1281
1282 // Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
1283 // produce partial content. When the partial message is text only we
1284 // keep whatever was streamed up to the cancel so the user does not
1285 // lose visible work (a cut-off summary is still useful). If the
1286 // partial message already contained tool-call blocks we drop the
1287 // whole thing, because an unmatched tool_use would fail the next
1288 // turn with a tool_result mismatch error.
1289 keep := len(finalMsg.Content) > 0
1290 if stop == provider.StopAborted && keep {
1291 hasToolCall := false
1292 for _, c := range finalMsg.Content {
1293 if _, ok := c.(provider.ToolCallBlock); ok {
1294 hasToolCall = true
1295 break
1296 }
1297 }
1298 if hasToolCall {
1299 keep = false
1300 }
1301 }
1302 if keep {
1303 emit := finalMsg
1304 suppress := false
1305
1306 // BeforeAssistantMessage hook: extensions can suppress or
1307 // rewrite the visible text. The transcript keeps the
1308 // model's original output so the model still sees what it
1309 // said on subsequent turns.
1310 if a.BeforeAssistantMessage != nil {
1311 orig := extractText(finalMsg)
1312 if orig != "" {
1313 allowed, _, replacement := a.BeforeAssistantMessage(orig)
1314 if !allowed {
1315 suppress = true
1316 } else if replacement != "" && replacement != orig {
1317 emit = replaceText(finalMsg, replacement)
1318 }
1319 }
1320 }
1321
1322 a.mu.Lock()
1323 a.messages = append(a.messages, finalMsg)
1324 a.rev++
1325 a.mu.Unlock()
1326 a.fireMessageAppended(finalMsg)
1327 if !suppress {
1328 sink(EvAssistantMessage{Message: emit})
1329 }
1330 // Now surface tool calls as EvToolCall events so UIs can render them
1331 // in order before the tool results arrive.
1332 for _, c := range finalMsg.Content {
1333 if tc, ok := c.(provider.ToolCallBlock); ok {
1334 sink(EvToolCall{ID: tc.ID, Name: tc.Name, Args: tc.Arguments})
1335 }
1336 }
1337 }
1338
1339 return stop, finalMsg, finalErr
1340}
1341
1342func drainProviderStream(stream <-chan provider.Event) {
1343 for range stream {
1344 }
1345}
1346
1347// ToolResultCommitError supplies a user-safe explanation when a host rejects
1348// a tool result during durable-state commit. Internal storage errors remain in
1349// Err; callers without this type retain the generic failure message.
1350type ToolResultCommitError struct {
1351 Message string
1352 Err error
1353}
1354
1355func (e *ToolResultCommitError) Error() string {
1356 if e == nil || e.Err == nil {
1357 return "tool result commit failed"
1358 }
1359 return e.Err.Error()
1360}
1361
1362func (e *ToolResultCommitError) Unwrap() error {
1363 if e == nil {
1364 return nil
1365 }
1366 return e.Err
1367}
1368
1369func toolResultCommitMessage(err error) string {
1370 var safe *ToolResultCommitError
1371 if errors.As(err, &safe) && strings.TrimSpace(safe.Message) != "" {
1372 return safe.Message
1373 }
1374 return "tool result state could not be persisted"
1375}
1376
1377// executeTools runs every tool call in the assistant message and returns
1378// a single tool-role message carrying all results.
1379func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink func(AgentEvent)) (provider.Message, bool) {
1380 var results []provider.Content
1381 var addedTools []string
1382 hadError := false
1383 tools := a.ToolsSnapshot()
1384
1385 for _, c := range msg.Content {
1386 tc, ok := c.(provider.ToolCallBlock)
1387 if !ok {
1388 continue
1389 }
1390 res := a.runOneTool(ctx, tc, tools, sink)
1391 // A plan tool result carries the parsed operation, not the resulting
1392 // checklist: core validates it against the current plan and materializes
1393 // what would be stored. Nothing mutates state here; that happens only
1394 // after CommitToolResult persisted the materialized list below.
1395 if op, ok := res.Details.(PlanOperation); ok && !res.IsError {
1396 update, err := a.previewPlanOperation(op)
1397 switch {
1398 case err != nil:
1399 // Validation failed: keep the current plan and surface the reason.
1400 res = ToolResult{
1401 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1402 IsError: true,
1403 Timing: res.Timing,
1404 }
1405 case op.Action == "show":
1406 // show is a read: return the listing without persisting a meta row
1407 // or committing anything.
1408 res.Content = []provider.Content{provider.TextBlock{Text: planResultText(op.Action, update)}}
1409 res.Details = nil
1410 res.PlanOperation = nil
1411 default:
1412 res.Details = update
1413 res.PlanOperation = &op
1414 res.Content = []provider.Content{provider.TextBlock{Text: planResultText(op.Action, update)}}
1415 }
1416 }
1417 if a.CommitToolResult != nil {
1418 if err := a.CommitToolResult(tc.ID, res); err != nil {
1419 res = ToolResult{
1420 Content: []provider.Content{provider.TextBlock{Text: toolResultCommitMessage(err)}},
1421 IsError: true,
1422 Timing: res.Timing,
1423 }
1424 }
1425 }
1426 if res.IsError {
1427 hadError = true
1428 }
1429 results = append(results, provider.ToolResultBlock{
1430 CallID: tc.ID,
1431 Content: res.Content,
1432 IsError: res.IsError,
1433 Timing: res.Timing,
1434 Context: res.Context,
1435 })
1436 for _, name := range res.ActivateTools {
1437 if _, err := tools.Get(name); err == nil && !containsString(addedTools, name) {
1438 addedTools = append(addedTools, name)
1439 }
1440 }
1441 if a.OnToolResult != nil {
1442 a.OnToolResult(tc.ID, res)
1443 }
1444 // Commit the validated plan only after the result was persisted: a failed
1445 // write replaced the result above and cleared PlanOperation, so state and
1446 // the event stream stay untouched.
1447 if update, ok := res.Details.(PlanUpdate); ok && !res.IsError && res.PlanOperation != nil {
1448 a.commitPlanUpdate(update)
1449 sink(EvPlanUpdate{CallID: tc.ID, Update: update})
1450 }
1451 sink(EvToolResult{ID: tc.ID, Result: res})
1452 }
1453
1454 return provider.Message{
1455 Role: provider.RoleTool,
1456 Content: results,
1457 Time: time.Now(),
1458 AddedToolNames: addedTools,
1459 }, hadError
1460}
1461
1462func (a *Agent) runOneTool(ctx context.Context, tc provider.ToolCallBlock, tools Registry, sink func(AgentEvent)) (res ToolResult) {
1463 started := time.Now()
1464 defer func() {
1465 if recover() != nil {
1466 res = ToolResult{
1467 Content: []provider.Content{provider.TextBlock{Text: "tool execution failed"}},
1468 IsError: true,
1469 }
1470 }
1471 res.Timing = &provider.ToolTiming{
1472 StartedAt: started,
1473 CompletedAt: time.Now(),
1474 Duration: time.Since(started),
1475 }
1476 }()
1477
1478 tool, err := tools.Get(tc.Name)
1479 if err != nil {
1480 return ToolResult{
1481 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1482 IsError: true,
1483 }
1484 }
1485
1486 args := tc.Arguments
1487
1488 // Intercept hook: an extension or other guard can refuse the
1489 // call before any side effect happens, OR rewrite the args
1490 // seen by the tool. The model sees the reason as the tool
1491 // error, learns from it, and (typically) proposes a different
1492 // action; rewrites are invisible to the model (they apply only
1493 // to the execution).
1494 if a.BeforeToolExecute != nil {
1495 allowed, reason, modified := a.BeforeToolExecute(tc)
1496 if !allowed {
1497 // A guard or confirmation denial is tracked separately from an
1498 // ordinary tool error: a denied invocation must not gain a
1499 // recovery prompt urging more actions.
1500 a.markDeniedToolCall()
1501 if reason == "" {
1502 reason = "tool call refused by extension guard"
1503 }
1504 return ToolResult{
1505 Content: []provider.Content{provider.TextBlock{Text: reason}},
1506 IsError: true,
1507 }
1508 }
1509 if len(modified) > 0 && json.Valid(modified) {
1510 if policy, ok := tool.(ToolArgumentRewritePolicy); ok && !policy.AllowArgumentRewrite() {
1511 return ToolResult{
1512 Content: []provider.Content{provider.TextBlock{Text: "tool arguments cannot be rewritten"}},
1513 IsError: true,
1514 }
1515 }
1516 args = modified
1517 }
1518 }
1519
1520 if len(args) == 0 {
1521 args = json.RawMessage("{}")
1522 }
1523
1524 // Tool panics are recovered by the invocation-level defer above so
1525 // guard, lookup, progress, and Execute panics all receive timing.
1526 sink(EvToolExecutionStarted{ID: tc.ID, Name: tc.Name})
1527 out, err := tool.Execute(ctx, args, func(text string) {
1528 sink(EvToolProgress{ID: tc.ID, Text: text})
1529 })
1530 if err != nil {
1531 // A refused execution (permission scope, allowlist, jail, or
1532 // admission gate) is tracked separately from an ordinary tool
1533 // error: like a guard or confirmation denial, it must not gain
1534 // a recovery prompt urging more actions.
1535 if asToolDeniedError(err) {
1536 a.markDeniedToolCall()
1537 }
1538 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
1539 return ToolResult{
1540 Content: []provider.Content{provider.TextBlock{Text: "aborted: " + err.Error()}},
1541 IsError: true,
1542 }
1543 }
1544 return ToolResult{
1545 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1546 IsError: true,
1547 }
1548 }
1549 return out
1550}
1551
1552// extractText concatenates all TextBlock content in a message. Used
1553// by BeforeAssistantMessage so guards see a single string instead of
1554// having to walk provider.Content themselves.
1555func mirrorToolImagesAsUser(msg provider.Message) provider.Message {
1556 var content []provider.Content
1557 for _, c := range msg.Content {
1558 tr, ok := c.(provider.ToolResultBlock)
1559 if !ok {
1560 continue
1561 }
1562 for _, inner := range tr.Content {
1563 switch v := inner.(type) {
1564 case provider.TextBlock:
1565 // Keep short textual context so the model understands why
1566 // the images appeared, but don't duplicate giant read
1567 // outputs verbatim.
1568 if len(v.Text) > 0 && len(v.Text) <= 500 {
1569 content = append(content, v)
1570 }
1571 case provider.ImageBlock:
1572 content = append(content, v)
1573 }
1574 }
1575 }
1576 if len(content) == 0 {
1577 return provider.Message{}
1578 }
1579 prefix := provider.TextBlock{Text: "Tool output included the following image content:"}
1580 content = append([]provider.Content{prefix}, content...)
1581 return provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
1582}
1583
1584const (
1585 maxTurnContextBytes = 16 * 1024
1586 turnContextTruncatedMarker = "\n[extension context truncated]"
1587)
1588
1589func boundedTurnContext(text string) string {
1590 text = strings.TrimSpace(text)
1591 if len(text) <= maxTurnContextBytes {
1592 return text
1593 }
1594 limit := maxTurnContextBytes - len(turnContextTruncatedMarker)
1595 cut := limit
1596 for cut > 0 && !utf8.RuneStart(text[cut]) {
1597 cut--
1598 }
1599 return strings.TrimSpace(text[:cut]) + turnContextTruncatedMarker
1600}
1601
1602func containsString(values []string, want string) bool {
1603 for _, value := range values {
1604 if value == want {
1605 return true
1606 }
1607 }
1608 return false
1609}
1610
1611func extractText(msg provider.Message) string {
1612 var out string
1613 for _, c := range msg.Content {
1614 if tb, ok := c.(provider.TextBlock); ok {
1615 if out != "" {
1616 out += "\n"
1617 }
1618 out += tb.Text
1619 }
1620 }
1621 return out
1622}
1623
1624// replaceText returns a copy of msg with every TextBlock replaced by
1625// a single TextBlock containing replacement. Non-text content (tool
1626// calls, etc.) is preserved in order.
1627func replaceText(msg provider.Message, replacement string) provider.Message {
1628 out := provider.Message{Role: msg.Role}
1629 out.Content = make([]provider.Content, 0, len(msg.Content))
1630 replaced := false
1631 for _, c := range msg.Content {
1632 if _, ok := c.(provider.TextBlock); ok {
1633 if !replaced {
1634 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1635 replaced = true
1636 }
1637 continue
1638 }
1639 out.Content = append(out.Content, c)
1640 }
1641 if !replaced {
1642 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1643 }
1644 return out
1645}