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 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
789 return false
790 }
791 msg := strings.ToLower(err.Error())
792 if msg == "" || isNonRetryableProviderLimit(msg) {
793 return false
794 }
795 if isStreamNetworkError(err, msg) {
796 return true
797 }
798 needles := []string{
799 "overloaded", "provider returned error", "rate limit", "ratelimit", "too many requests",
800 "429", "http 429", "500", "http 500", "502", "http 502", "503", "http 503", "504", "http 504",
801 "service unavailable", "server error", "internal error", "ended without", "stream ended before",
802 "did not get a response", "timed out", "timeout", "terminated",
803 // OpenAI's ChatGPT/Codex backend returns this generic message (with a
804 // request ID) for transient server failures and explicitly says
805 // "You can retry your request".
806 "an error occurred while processing your request",
807 // Explicit retry guidance emitted by provider backends (OpenAI
808 // Responses, AWS Bedrock stream exceptions) with varying prefixes.
809 "you can retry your request", "try your request again", "please retry your request",
810 // Capacity messages from the ChatGPT/Codex backend, e.g.
811 // "Our servers are currently overloaded. Please try again later."
812 // The trailing advice also shows up on its own for transient
813 // capacity failures; usage/quota limits are filtered out above by
814 // isNonRetryableProviderLimit before this list is consulted.
815 "servers are currently overloaded", "servers are busy", "try again later",
816 }
817 for _, needle := range needles {
818 if strings.Contains(msg, needle) {
819 return true
820 }
821 }
822 return false
823}
824
825// isStreamNetworkError detects transport errors from an opened provider stream.
826// Keep typed EOF checks because WebSocket readers return io.EOF on a peer close,
827// while provider-specific transports may only preserve a textual error.
828func isStreamNetworkError(err error, msg string) bool {
829 if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
830 return true
831 }
832 for _, needle := range []string{
833 "connection", "network", "unexpected eof", "broken pipe", "socket hang up",
834 "transport failure", "upstream connect",
835 } {
836 if strings.Contains(msg, needle) {
837 return true
838 }
839 }
840 return false
841}
842
843func isNonRetryableProviderLimit(msg string) bool {
844 needles := []string{
845 "usage limit", "monthly usage limit", "freeusagelimit", "gousagelimit",
846 "available balance", "insufficient_quota", "out of budget", "quota exceeded", "billing",
847 }
848 for _, needle := range needles {
849 if strings.Contains(msg, needle) {
850 return true
851 }
852 }
853 return false
854}
855
856func classifyRetryReason(err error) RetryReason {
857 if err == nil {
858 return RetryReasonUnknown
859 }
860 msg := strings.ToLower(err.Error())
861 if isNonRetryableProviderLimit(msg) {
862 return RetryReasonQuota
863 }
864 if strings.Contains(msg, "overload") || strings.Contains(msg, "servers are busy") {
865 return RetryReasonOverload
866 }
867 if strings.Contains(msg, "rate limit") || strings.Contains(msg, "ratelimit") || strings.Contains(msg, "too many requests") || strings.Contains(msg, "429") {
868 return RetryReasonRateLimit
869 }
870 if errors.Is(err, context.DeadlineExceeded) || strings.Contains(msg, "timeout") || strings.Contains(msg, "timed out") {
871 return RetryReasonTimeout
872 }
873 if isStreamNetworkError(err, msg) {
874 return RetryReasonNetwork
875 }
876 for _, needle := range []string{"context window", "context length", "maximum context", "too many tokens"} {
877 if strings.Contains(msg, needle) {
878 return RetryReasonContextWindow
879 }
880 }
881 for _, needle := range []string{"unauthorized", "authentication", "invalid api key", "invalid_api_key", "http 401", "http 403"} {
882 if strings.Contains(msg, needle) {
883 return RetryReasonAuth
884 }
885 }
886 for _, needle := range []string{"server error", "internal error", "service unavailable", "http 500", "http 502", "http 503", "http 504", "provider returned error"} {
887 if strings.Contains(msg, needle) {
888 return RetryReasonServer
889 }
890 }
891 for _, needle := range []string{"bad request", "unsupported", "invalid parameter", "http 400", "http 404", "http 422"} {
892 if strings.Contains(msg, needle) {
893 return RetryReasonClient
894 }
895 }
896 return RetryReasonUnknown
897}
898
899func (a *Agent) fireRetryLifecycle(record RetryLifecycleRecord) {
900 if a.OnRetryLifecycle != nil {
901 a.OnRetryLifecycle(record)
902 }
903}
904
905func (a *Agent) retryDelay(attempt int) time.Duration {
906 base := a.RetryBaseDelay
907 if base <= 0 {
908 base = 2 * time.Second
909 }
910 return base * time.Duration(1<<attempt)
911}
912
913func sleepRetry(ctx context.Context, d time.Duration) error {
914 if d <= 0 {
915 return ctx.Err()
916 }
917 t := time.NewTimer(d)
918 defer t.Stop()
919 select {
920 case <-ctx.Done():
921 return ctx.Err()
922 case <-t.C:
923 return nil
924 }
925}
926
927func (a *Agent) dropLastAssistantMessage() {
928 a.mu.Lock()
929 defer a.mu.Unlock()
930 if n := len(a.messages); n > 0 && a.messages[n-1].Role == provider.RoleAssistant {
931 a.messages = a.messages[:n-1]
932 a.rev++
933 }
934}
935
936// requestLifecycleSink adapts provider-owned pre-stream retries into core
937// events without letting provider concerns leak into mode renderers.
938type requestLifecycleSink struct {
939 sink func(AgentEvent)
940 observe func(RetryLifecycleRecord)
941 provider string
942 model string
943 lastReason RetryReason
944}
945
946func (s *requestLifecycleSink) CacheDiagnostics(diagnostics provider.CacheDiagnostics) {
947 s.sink(EvCacheDiagnostics{
948 Eligible: diagnostics.Eligible,
949 Mode: diagnostics.Mode,
950 Transport: diagnostics.Transport,
951 Continuation: diagnostics.Continuation,
952 })
953}
954
955func (s *requestLifecycleSink) RequestAttempt(attempt, maxAttempts int) {
956 s.sink(EvRequestStarted{
957 Provider: s.provider,
958 Model: s.model,
959 Scope: RetryScopeProvider,
960 Attempt: attempt,
961 MaxAttempts: maxAttempts,
962 })
963}
964
965func (s *requestLifecycleSink) RequestFailed(attempt, maxAttempts int, reason provider.RequestFailureReason, terminal bool) {
966 s.lastReason = retryReasonFromProvider(reason)
967 if s.observe != nil {
968 s.observe(RetryLifecycleRecord{
969 Event: RetryLifecycleRequestFailed,
970 Scope: RetryScopeProvider,
971 Attempt: attempt,
972 MaxAttempts: maxAttempts,
973 Reason: s.lastReason,
974 Terminal: terminal,
975 })
976 }
977}
978
979func (s *requestLifecycleSink) RetryScheduled(attempt, maxAttempts int, delay time.Duration) {
980 s.sink(EvRetryScheduled{
981 Scope: RetryScopeProvider,
982 Attempt: attempt,
983 MaxAttempts: maxAttempts,
984 Delay: delay,
985 })
986 if s.observe != nil {
987 reason := s.lastReason
988 if reason == "" {
989 reason = RetryReasonUnknown
990 }
991 s.observe(RetryLifecycleRecord{
992 Event: RetryLifecycleRetryScheduled,
993 Scope: RetryScopeProvider,
994 Attempt: attempt,
995 MaxAttempts: maxAttempts,
996 Reason: reason,
997 DelayMS: delay.Milliseconds(),
998 })
999 }
1000 s.lastReason = ""
1001}
1002
1003func retryReasonFromProvider(reason provider.RequestFailureReason) RetryReason {
1004 switch reason {
1005 case provider.RequestFailureOverload:
1006 return RetryReasonOverload
1007 case provider.RequestFailureRateLimit:
1008 return RetryReasonRateLimit
1009 case provider.RequestFailureQuota:
1010 return RetryReasonQuota
1011 case provider.RequestFailureServer:
1012 return RetryReasonServer
1013 case provider.RequestFailureNetwork:
1014 return RetryReasonNetwork
1015 case provider.RequestFailureTimeout:
1016 return RetryReasonTimeout
1017 case provider.RequestFailureClient:
1018 return RetryReasonClient
1019 default:
1020 return RetryReasonUnknown
1021 }
1022}
1023
1024// oneTurn calls the LLM once, forwards events, returns the stop reason
1025// and the assembled assistant message (already appended to the transcript).
1026func newStreamIdleTimer(timeout time.Duration) (<-chan time.Time, func(), func()) {
1027 timer := time.NewTimer(timeout)
1028 stop := func() { timer.Stop() }
1029 reset := func() {
1030 if !timer.Stop() {
1031 select {
1032 case <-timer.C:
1033 default:
1034 }
1035 }
1036 timer.Reset(timeout)
1037 }
1038 return timer.C, stop, reset
1039}
1040
1041func (a *Agent) oneTurn(ctx context.Context, sink func(AgentEvent), turnContext string, requestContext provider.RequestContext, attempt, maxAttempts int) (provider.StopReason, provider.Message, error) {
1042 fastMode := a.FastModeEnabled()
1043 if err := provider.ValidateFastMode(a.Client.Name(), fastMode); err != nil {
1044 return provider.StopError, provider.Message{}, err
1045 }
1046 system, tools := a.PromptConfig()
1047 // Repair pairs before projecting the copied provider-input view. The
1048 // repair can add stub results for aborted calls; those results must also
1049 // remain in the request so every tool call still has a matching result.
1050 messages := repairToolUseResultPairs(a.Messages())
1051 requestLifecycle := &requestLifecycleSink{
1052 sink: sink,
1053 observe: a.fireRetryLifecycle,
1054 provider: a.Client.Name(),
1055 model: a.Model,
1056 }
1057 req := provider.Request{
1058 Model: a.Model,
1059 System: system,
1060 // Repair any dangling tool_use blocks before sending. A turn
1061 // aborted mid-flight (cancel, connection drop, ECONNREFUSED to a
1062 // dev server, etc.) can leave an assistant tool_use with no
1063 // matching tool_result in the live transcript. The load-time
1064 // repair in OpenSession only runs on restart, so without this the
1065 // next in-process request is rejected by providers like Anthropic
1066 // with "tool_use ids were found without tool_result blocks". The
1067 // repair is pure and a no-op on already-valid transcripts.
1068 Messages: projectProviderMessages(messages),
1069 Tools: tools.Specs(),
1070 Reasoning: a.Reasoning,
1071 FastMode: fastMode,
1072 MaxTokens: a.MaxTokens,
1073 Temperature: a.Temperature,
1074 Lifecycle: requestLifecycle,
1075 Context: requestContext,
1076 }
1077 sink(EvRequestStarted{
1078 Provider: a.Client.Name(),
1079 Model: a.Model,
1080 Scope: RetryScopeAgent,
1081 Attempt: attempt + 1,
1082 MaxAttempts: maxAttempts,
1083 })
1084 streamCtx, cancelStream := context.WithCancel(ctx)
1085 defer cancelStream()
1086 a.beginProviderRequest()
1087 stream, err := a.Client.Stream(streamCtx, req)
1088 if err != nil {
1089 return provider.StopError, provider.Message{}, &requestOpenError{err: err}
1090 }
1091 // A provider may already have queued events when cancellation wins. Keep a
1092 // consumer attached after this turn returns so bounded producer channels
1093 // cannot strand the provider goroutine and its response body.
1094 defer func() { go drainProviderStream(stream) }()
1095
1096 sink(EvAssistantStart{})
1097
1098 var (
1099 stop provider.StopReason
1100 finalErr error
1101 finalMsg provider.Message
1102 )
1103
1104 var idle <-chan time.Time
1105 var resetIdle func()
1106 if a.StreamIdleTimeout > 0 {
1107 newTimer := a.streamIdleTimer
1108 if newTimer == nil {
1109 newTimer = newStreamIdleTimer
1110 }
1111 var stopIdle func()
1112 idle, stopIdle, resetIdle = newTimer(a.StreamIdleTimeout)
1113 defer stopIdle()
1114 }
1115 handleStreamEvent := func(ev provider.Event) (done bool) {
1116 if resetIdle != nil {
1117 resetIdle()
1118 }
1119 switch e := ev.(type) {
1120 case provider.EventStart:
1121 // nothing
1122 case provider.EventTextDelta:
1123 sink(EvTextDelta{Delta: e.Delta})
1124 case provider.EventToolStart:
1125 sink(EvToolUseStart{ID: e.ID, Name: e.Name})
1126 case provider.EventToolArgs:
1127 sink(EvToolUseArgs{ID: e.ID, Delta: e.Delta})
1128 case provider.EventToolEnd:
1129 sink(EvToolUseEnd{ID: e.ID})
1130 case provider.EventUsage:
1131 cum := a.addUsage(e.Usage)
1132 sink(EvUsage{Usage: e.Usage, Cumulative: cum})
1133 if a.OnUsage != nil {
1134 a.OnUsage(cum)
1135 }
1136 case provider.EventDone:
1137 stop = e.Stop
1138 finalErr = e.Err
1139 finalMsg = e.Message
1140 return true
1141 }
1142 return false
1143 }
1144 for {
1145 // Consume no more than one buffered event before honoring cancellation.
1146 select {
1147 case ev, ok := <-stream:
1148 if !ok || handleStreamEvent(ev) {
1149 goto streamDone
1150 }
1151 default:
1152 }
1153 if err := streamCtx.Err(); err != nil {
1154 return provider.StopError, finalMsg, err
1155 }
1156 select {
1157 case <-streamCtx.Done():
1158 return provider.StopError, finalMsg, streamCtx.Err()
1159 case <-idle:
1160 return provider.StopError, finalMsg, ErrStreamIdleTimeout
1161 case ev, ok := <-stream:
1162 if !ok || handleStreamEvent(ev) {
1163 goto streamDone
1164 }
1165 }
1166 }
1167streamDone:
1168
1169 // Append assistant message to transcript. Aborted turns (Esc / Ctrl+C)
1170 // produce partial content. When the partial message is text only we
1171 // keep whatever was streamed up to the cancel so the user does not
1172 // lose visible work (a cut-off summary is still useful). If the
1173 // partial message already contained tool-call blocks we drop the
1174 // whole thing, because an unmatched tool_use would fail the next
1175 // turn with a tool_result mismatch error.
1176 keep := len(finalMsg.Content) > 0
1177 if stop == provider.StopAborted && keep {
1178 hasToolCall := false
1179 for _, c := range finalMsg.Content {
1180 if _, ok := c.(provider.ToolCallBlock); ok {
1181 hasToolCall = true
1182 break
1183 }
1184 }
1185 if hasToolCall {
1186 keep = false
1187 }
1188 }
1189 if keep {
1190 emit := finalMsg
1191 suppress := false
1192
1193 // BeforeAssistantMessage hook: extensions can suppress or
1194 // rewrite the visible text. The transcript keeps the
1195 // model's original output so the model still sees what it
1196 // said on subsequent turns.
1197 if a.BeforeAssistantMessage != nil {
1198 orig := extractText(finalMsg)
1199 if orig != "" {
1200 allowed, _, replacement := a.BeforeAssistantMessage(orig)
1201 if !allowed {
1202 suppress = true
1203 } else if replacement != "" && replacement != orig {
1204 emit = replaceText(finalMsg, replacement)
1205 }
1206 }
1207 }
1208
1209 a.mu.Lock()
1210 a.messages = append(a.messages, finalMsg)
1211 a.rev++
1212 a.mu.Unlock()
1213 a.fireMessageAppended(finalMsg)
1214 if !suppress {
1215 sink(EvAssistantMessage{Message: emit})
1216 }
1217 // Now surface tool calls as EvToolCall events so UIs can render them
1218 // in order before the tool results arrive.
1219 for _, c := range finalMsg.Content {
1220 if tc, ok := c.(provider.ToolCallBlock); ok {
1221 sink(EvToolCall{ID: tc.ID, Name: tc.Name, Args: tc.Arguments})
1222 }
1223 }
1224 }
1225
1226 return stop, finalMsg, finalErr
1227}
1228
1229func drainProviderStream(stream <-chan provider.Event) {
1230 for range stream {
1231 }
1232}
1233
1234// ToolResultCommitError supplies a user-safe explanation when a host rejects
1235// a tool result during durable-state commit. Internal storage errors remain in
1236// Err; callers without this type retain the generic failure message.
1237type ToolResultCommitError struct {
1238 Message string
1239 Err error
1240}
1241
1242func (e *ToolResultCommitError) Error() string {
1243 if e == nil || e.Err == nil {
1244 return "tool result commit failed"
1245 }
1246 return e.Err.Error()
1247}
1248
1249func (e *ToolResultCommitError) Unwrap() error {
1250 if e == nil {
1251 return nil
1252 }
1253 return e.Err
1254}
1255
1256func toolResultCommitMessage(err error) string {
1257 var safe *ToolResultCommitError
1258 if errors.As(err, &safe) && strings.TrimSpace(safe.Message) != "" {
1259 return safe.Message
1260 }
1261 return "tool result state could not be persisted"
1262}
1263
1264// executeTools runs every tool call in the assistant message and returns
1265// a single tool-role message carrying all results.
1266func (a *Agent) executeTools(ctx context.Context, msg provider.Message, sink func(AgentEvent)) (provider.Message, bool) {
1267 var results []provider.Content
1268 var addedTools []string
1269 hadError := false
1270 tools := a.ToolsSnapshot()
1271
1272 for _, c := range msg.Content {
1273 tc, ok := c.(provider.ToolCallBlock)
1274 if !ok {
1275 continue
1276 }
1277 res := a.runOneTool(ctx, tc, tools, sink)
1278 if a.CommitToolResult != nil {
1279 if err := a.CommitToolResult(tc.ID, res); err != nil {
1280 res = ToolResult{
1281 Content: []provider.Content{provider.TextBlock{Text: toolResultCommitMessage(err)}},
1282 IsError: true,
1283 Timing: res.Timing,
1284 }
1285 }
1286 }
1287 if res.IsError {
1288 hadError = true
1289 }
1290 results = append(results, provider.ToolResultBlock{
1291 CallID: tc.ID,
1292 Content: res.Content,
1293 IsError: res.IsError,
1294 Timing: res.Timing,
1295 })
1296 for _, name := range res.ActivateTools {
1297 if _, err := tools.Get(name); err == nil && !containsString(addedTools, name) {
1298 addedTools = append(addedTools, name)
1299 }
1300 }
1301 if a.OnToolResult != nil {
1302 a.OnToolResult(tc.ID, res)
1303 }
1304 if update, ok := res.Details.(PlanUpdate); ok && !res.IsError {
1305 sink(EvPlanUpdate{Update: update})
1306 }
1307 sink(EvToolResult{ID: tc.ID, Result: res})
1308 }
1309
1310 return provider.Message{
1311 Role: provider.RoleTool,
1312 Content: results,
1313 Time: time.Now(),
1314 AddedToolNames: addedTools,
1315 }, hadError
1316}
1317
1318func (a *Agent) runOneTool(ctx context.Context, tc provider.ToolCallBlock, tools Registry, sink func(AgentEvent)) (res ToolResult) {
1319 started := time.Now()
1320 defer func() {
1321 if recover() != nil {
1322 res = ToolResult{
1323 Content: []provider.Content{provider.TextBlock{Text: "tool execution failed"}},
1324 IsError: true,
1325 }
1326 }
1327 res.Timing = &provider.ToolTiming{
1328 StartedAt: started,
1329 CompletedAt: time.Now(),
1330 Duration: time.Since(started),
1331 }
1332 }()
1333
1334 tool, err := tools.Get(tc.Name)
1335 if err != nil {
1336 return ToolResult{
1337 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1338 IsError: true,
1339 }
1340 }
1341
1342 args := tc.Arguments
1343
1344 // Intercept hook: an extension or other guard can refuse the
1345 // call before any side effect happens, OR rewrite the args
1346 // seen by the tool. The model sees the reason as the tool
1347 // error, learns from it, and (typically) proposes a different
1348 // action; rewrites are invisible to the model (they apply only
1349 // to the execution).
1350 if a.BeforeToolExecute != nil {
1351 allowed, reason, modified := a.BeforeToolExecute(tc)
1352 if !allowed {
1353 if reason == "" {
1354 reason = "tool call refused by extension guard"
1355 }
1356 return ToolResult{
1357 Content: []provider.Content{provider.TextBlock{Text: reason}},
1358 IsError: true,
1359 }
1360 }
1361 if len(modified) > 0 && json.Valid(modified) {
1362 if policy, ok := tool.(ToolArgumentRewritePolicy); ok && !policy.AllowArgumentRewrite() {
1363 return ToolResult{
1364 Content: []provider.Content{provider.TextBlock{Text: "tool arguments cannot be rewritten"}},
1365 IsError: true,
1366 }
1367 }
1368 args = modified
1369 }
1370 }
1371
1372 if len(args) == 0 {
1373 args = json.RawMessage("{}")
1374 }
1375
1376 // Tool panics are recovered by the invocation-level defer above so
1377 // guard, lookup, progress, and Execute panics all receive timing.
1378 sink(EvToolExecutionStarted{ID: tc.ID, Name: tc.Name})
1379 out, err := tool.Execute(ctx, args, func(text string) {
1380 sink(EvToolProgress{ID: tc.ID, Text: text})
1381 })
1382 if err != nil {
1383 if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
1384 return ToolResult{
1385 Content: []provider.Content{provider.TextBlock{Text: "aborted: " + err.Error()}},
1386 IsError: true,
1387 }
1388 }
1389 return ToolResult{
1390 Content: []provider.Content{provider.TextBlock{Text: err.Error()}},
1391 IsError: true,
1392 }
1393 }
1394 return out
1395}
1396
1397// extractText concatenates all TextBlock content in a message. Used
1398// by BeforeAssistantMessage so guards see a single string instead of
1399// having to walk provider.Content themselves.
1400func mirrorToolImagesAsUser(msg provider.Message) provider.Message {
1401 var content []provider.Content
1402 for _, c := range msg.Content {
1403 tr, ok := c.(provider.ToolResultBlock)
1404 if !ok {
1405 continue
1406 }
1407 for _, inner := range tr.Content {
1408 switch v := inner.(type) {
1409 case provider.TextBlock:
1410 // Keep short textual context so the model understands why
1411 // the images appeared, but don't duplicate giant read
1412 // outputs verbatim.
1413 if len(v.Text) > 0 && len(v.Text) <= 500 {
1414 content = append(content, v)
1415 }
1416 case provider.ImageBlock:
1417 content = append(content, v)
1418 }
1419 }
1420 }
1421 if len(content) == 0 {
1422 return provider.Message{}
1423 }
1424 prefix := provider.TextBlock{Text: "Tool output included the following image content:"}
1425 content = append([]provider.Content{prefix}, content...)
1426 return provider.Message{Role: provider.RoleUser, Content: content, Time: time.Now()}
1427}
1428
1429const (
1430 maxTurnContextBytes = 16 * 1024
1431 turnContextTruncatedMarker = "\n[extension context truncated]"
1432)
1433
1434func boundedTurnContext(text string) string {
1435 text = strings.TrimSpace(text)
1436 if len(text) <= maxTurnContextBytes {
1437 return text
1438 }
1439 limit := maxTurnContextBytes - len(turnContextTruncatedMarker)
1440 cut := limit
1441 for cut > 0 && !utf8.RuneStart(text[cut]) {
1442 cut--
1443 }
1444 return strings.TrimSpace(text[:cut]) + turnContextTruncatedMarker
1445}
1446
1447func containsString(values []string, want string) bool {
1448 for _, value := range values {
1449 if value == want {
1450 return true
1451 }
1452 }
1453 return false
1454}
1455
1456func extractText(msg provider.Message) string {
1457 var out string
1458 for _, c := range msg.Content {
1459 if tb, ok := c.(provider.TextBlock); ok {
1460 if out != "" {
1461 out += "\n"
1462 }
1463 out += tb.Text
1464 }
1465 }
1466 return out
1467}
1468
1469// replaceText returns a copy of msg with every TextBlock replaced by
1470// a single TextBlock containing replacement. Non-text content (tool
1471// calls, etc.) is preserved in order.
1472func replaceText(msg provider.Message, replacement string) provider.Message {
1473 out := provider.Message{Role: msg.Role}
1474 out.Content = make([]provider.Content, 0, len(msg.Content))
1475 replaced := false
1476 for _, c := range msg.Content {
1477 if _, ok := c.(provider.TextBlock); ok {
1478 if !replaced {
1479 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1480 replaced = true
1481 }
1482 continue
1483 }
1484 out.Content = append(out.Content, c)
1485 }
1486 if !replaced {
1487 out.Content = append(out.Content, provider.TextBlock{Text: replacement})
1488 }
1489 return out
1490}