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