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