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