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